package org.NooLab.glue;


import java.text.DecimalFormat;
import java.util.*;

import org.NooLab.utilities.EncounterCounter;
import org.NooLab.utilities.files.DFutils;
import org.NooLab.utilities.logging.PrintLog;
import org.NooLab.utilities.net.GUID;
import org.NooLab.utilities.net.connex.NicAddresses;
import org.NooLab.utilities.nums.NumUtilities;
import org.NooLab.utilities.strings.ArrUtilities;
import org.NooLab.utilities.strings.StringsUtil;
import org.NooLab.utilities.timing.DelayFor;

 
import org.NooLab.glue.components.*;
import org.NooLab.glue.filter.*;

import org.NooLab.glue.msgframe.*;
import org.NooLab.glue.net.*;
import org.NooLab.glue.transaction.*;

import org.NooLab.glue.net.env.ConnectionsManager;
import org.NooLab.glue.recdispatcher.ControlTaskHandler;
import org.NooLab.glue.recdispatcher.ReceptorLinks;
import org.NooLab.glue.storage.MsgBoardDataStorage;
import org.NooLab.glue.subscription.ActiveContextHeaderDescriptions;
import org.NooLab.glue.subscription.FilterPackageIntf;
import org.NooLab.glue.subscription.FilterPackages;
import org.NooLab.glue.subscription.Futures;
import org.NooLab.glue.subscription.FuturesIntf;
import org.NooLab.glue.subscription.MessageMatcher;
import org.NooLab.glue.subscription.MessageMatcherProcess;



/**
 * 
 * explains how to embedding Jetty -> is able to provide WebSockets,
 * such the connectivity via http could be the same as for local TCP !
 * 
 * http://angelozerr.wordpress.com/2011/07/26/websockets_jetty_step3/
 * 
 * 
 * 
 * http://php-java-bridge.sourceforge.net/pjb/how_it_works.php
 * 
 * http://tutorials.jenkov.com/java-generics/class-objects-as-type-literals.htmliboardMessagesconn
 * 
 * Driver employee   = read(Driver.class, "select * from drivers where id=1");
 * Vehicle vehicle   = read(Vehicle.class, "select * from vehicles where id=1");
 * 
 * returning instance of type defined by the class-literal
 * public static <T> T read(Class<T> theClass, String sql)
 * 		T o = theClass.newInstance();	
 * 		return o;
 * }
 * 
 * see http://mom4j.sourceforge.net/
 * 
 * 
 * 
 * 
 * interesting stuff @ protocol7
 * 
 * https://github.com/protocol7/rest-demo/tree/master/rest-server/src/main/java/com/protocol7/demo/rest/server
 *   
 *     
 *       // ATTENTION notification ID is NOT always == transaction ID !!
 *    
 *    
 *    
 * TODO  messages in non-transactional mode
 *       advantage     : much faster
 *       disadvantages : - connections are not ensured
 *                       - no centrally issued message id, hence  
 *                         - no explicit alignment, 
 *                         - reduced error handling,
 *                         - no persistence of transactions  
 *    
 *    
 * TODO  relaying of messages by subscription instead of / additionally to doc type and group
 *       subscription should respect timeout of transactions   
 *    
 * TODO  running the MessageBoard in dual-protocol-mode: local clients connected through TCP, remote board through http-rest
 *    
 * TODO  remote tcp protocol: if http is available, tcp should be available, too
 *    
 *     
 * TODO  need a refreshing & assigning process for available RECEPTORS    
 *     
 *    
 * TODO  Supporting TTL (time to live) per message, per registration
 *       first send of an unregistered client -> register on the fly !
 *        
 *     
 * TODO  participants registering with different roles, or a bivalent role, 
 * 		 -> done, but not tested, it is controlled by a variable in the factory properties "instanceBivalence" (boolean) 
 *   
 * TODO  provide the possibility for aligned results; fully ordered, semi-ordered , as-soon-as-possible
 *       in many tasks this is necessary to keep a semantic balance 
 * 		 - if requested, the results should be delivered in the "same" order as the tasks
 *       - define mandatory pillars (resulting in a semi-ordered delivery)  
 *         either explicitly or as a "basked size"    
 *     
 *       it is simply achieved by an additional layer for accepting tasks (registering the order)
 *       and a procedural layer establishing the order
 *       
 *       alignedResults = 1=ASAP/2=SEMI/3=FULL; semiAlignedResultsBasketSize=4, 
 *       
 *       this property is a property of a source client, which receives results
 *       the mechanism is simply buffering and sorting in the MessageBoard 
 *       
 *       => if a milestone step fails, all steps until the milestone also fail ! == will not be delivered
 *       
 *     
 * TODO  board-to-board communication
 *       introduce control for hop counts in relay into source message    
 * 
 *   
 * TODO  context definitions may refer 
 *        - to the format (xml tags)  (often defined through filters)
 *        - to the content (String) 
 *        - to filters
 *        - document type descriptors (as String) (very simple)
 *       or any combination 
 *                
 * 
 * TODO: CRC32 of the message to be contained IN the message (as the last xml tag)
 *       return an err message to any sender if the received message is not consistent
 *       
 *       checking the completeness for any acceptance
 * 
 * 
 * TODO  acting as a "ConnexLinkServer", if requested/allowed by the participants, AND by the message
 *       Yet, this works only in local networks
 * 
 * TODO  a streaming mode for high frequency transmission of small packets
 *       without confirmations etc. (video streaming)
 *       but with a byte oriented buffer , 
 *       essentially mapping streams onto each other on the level of bytes
 *       1:n connections are allowed (diverting streams)
 * 		 "frequency" and size of packets are parameters
 * 
 *       the trick is to use
 *       1- a connection GUID as an abbreviation for the XML stuff
 *       2- using this GUID in its byte-form as a non-xml "tag", i.e. as a header of each (byte) message  
 * 		 
 * 	     such, no hand-shake will take place, the MessageBoard knows by means of the GUID 
 * 	     how to route the message (from RECEPTOR to SOURCE)
 *   
 *       note, that the data source is on the side of the SOURCE (e.g.a webcam )
 *       this works also for RESTful connections, where a special channel is opened for high-frequency "get"s,
 *       such mimicking a "stream"
 *       
 * 
 *                   
 * TODO  virtual ports = tag inside a message and intra-board relay                    
 *       as for physical ports, this is to distinguish "active workers" from mere clients (like iPhones)  
 *       within registration, the MessageBoard tells the client whether there
 *       are separate ports or if there is a single port with virtual ports
 *       This would allow for using a single port even for direct transport protocols
 *         
 * 
 *         
 *         
 */
public class MessageBoard  implements Observer{

	// =================================

	public boolean DEBUG=false; // true=prevents timing out of connections and shutting down the client application,
							   // else, no complete connection check is performed
	
	// constants ......................
	
	/** 1=active sending possible, 
	 *  2=using http-rest, confirmations are returned in the handshake, but results/data need to be polled by the clients
	 *  3=the board is completely passive, even confirmations need to be polled (not implemented yet)
     */
	int sendModeForBoard = 1;
	/** the restlet mode handles this transparently on its own! */
	private boolean virtualPorts ;
	
	int lowestPortNumber  = 7300;
	int highestPortNumber = 9000;
	
	/** the standard return port BEFORE registration, used by the MessageBoard to return the final port assignment */
	int msgPortforRegistration = 7112 ;
	
	int httpServerPort      = 6100 ;
	
	int msgPortAllInOne     = 7070;
	
	int msgPortforInterBoard = 7007;
	int msgPortforSources   = 7100;
	int msgPortforReceptors = 7200;
	
	int msgPortforService    = 7111;
	 
	// default value, yet, each participant may provide its own channel via registration

	ArrayList<String> ipAddressList ; 
	String ipAddressListStr="" ;
	HostingData hostingData = new HostingData() ;
		
	public static double connectionDelayDefault = 1.0 ;
	
	/** this is per client */
	int maxNumberOfBufferedTasks = 50;
	int dropDelayUnrelayedMessages = 1;
	
	// object references ..............
	
	MessageBoardShareIntf msgBoardShares;
	MessageBoardFactory mbf ;
	
	GlobalReceipts globalReceipts ; 
		
	ComBoxFactory comBoxFactory ;
	ComBoxIntf comBox ;

	/** the only place where new transactions are added is in method "sourceVariant_RequestForTransactionReceipt()",
	 *  i.e. if a participant requests an ID (such taking the role of a SOURCE), the transaction slot is opened.
	 *  This remains alive (by internal state of the transaction, and as an item in the collection of transactions), 
	 *  until the results have been delivered (or, in the case of "travel" messages, the msg has been passed through)
	 *  
	 *  like the notifications, the wrapper object cares itself for loading on construction;
	 *  saving, however, is organized differently: transactions are save upon "stopMessageBoard()", 
	 *  after adding, removing an item, and after relaying the messages (to clients of any kind)
	 */
	Transactions transactions ;
	
	MessageMatcher  messageMatcher ;
	MessageMatcherProcess msgmatcherProcess;
	FuturesIntf futures; // we will not create it here, instead, we get it from the factory
	
	
	private MsgBoardDataStorage msgBoardStorage ;
	
	RegisterMessageInterpreter registerMessageInterpreter ;
	SourceMessageInterpreter sourceMessageInterpreter ;
	ReceptorMessageInterpreter receptorMessageInterpreter;
	BoardMessageInterpreter boardMessageInterpreter;
	

	// all messages coming in from sources
	protected Vector<Message> sourceMessages = new Vector<Message>();
	// all messages coming in from receptors
	protected Vector<Message> receptorMessages = new Vector<Message>();
	
	protected Vector<Message> registerMessages = new Vector<Message>();
	
	
	// all messages prepared
	protected Vector<Message> boardMessages = new Vector<Message>();

	private RegisterHandler registerHandler;
	private SourceHandler srcHandler;
	private ClientHandler cliHandler;
	private BoardMessageHandler mbHandler;
	private InBufferHandler inBuffHandler;
	private NotificationHandler notifyHandler;
	
	private TaskRelayDispatcherProcess taskRelayDispatcherProcess;
	private MonitorProcess monitorProcess;
	TaskItemsCleanupCollector cleanupCollector;
	
	BoardAddresses boardAddresses=null ;
	
	ReceptorLinks receptorLinks=null;
	
	MessageBoardTrafficMonitorIntf  monitor;
	
	ClientMatcherIntf clientMatcher;
	
	Notifications waitingNotifications ;    
	int notificationIdlingValue=0;
	
	ConnectionsManager connectionsMgr;
		
	MetaRegistration metaRegistration;
	
	RegistrationManager registrationMgr;
	long registrationTimeOut;
	int modeForConflictingIDs = 2;
	
	protected Vector<Long> deliveredResultsTID = new Vector<Long>();
	
	protected Vector<Long> instancesForResultAlignment = new Vector<Long>();
	
	ActiveContextHeaderDescriptions activeContextHeaderDescriptions;
	
	double avgReceptorsReCheckTimevalue = 25.0 ;
	int standardPeriodForConnectionCheck = 1400;
	int periodLengthForTaskDispatcher = 8000;

	
	/**
	 * FilterPackages are just a convenient class for collections of filter-s;
	 * filters in FilterPackages are implemented into the MessageBoard through a particular
	 * group of transactions;
	 */
	private FilterPackages filters = new FilterPackages ();
	
	
	@SuppressWarnings("unused")
	private DataReceiverIntf  basicCliReceiver, basicRegistrar, uniformHttpReceiver ;
	
	// such direct instances should not be seen here !!
	// private UdpDataReceiver  basicCliReceiver0,basicSrcReceiver0, basicRegistrar0 ;
	// private UdpDataReceiver  testReceiver; 
	// private UdpDataSender  mbSender0;
	// we should just operate on the level of the ComBoxIntf !! 
	
	
	protected Vector<Object> inBuffer = new Vector<Object>(); 
	
	// main variables / properties ....

	String configHomePath="" ;
	String homePath="" ;

	String simpleName = "mb";
	String messageBoardsIdentifier = "";
	XRegistration mxReg;
	 
	
	/** will be 0 for all transports <= ftp, but can be customized actively also for UDP
	 *  transport will be accessed through factoryProperties in the factory 
	 */
	int fullDuplexConnection = 0 ;
	
	int protocolID = -1; // that's just for an abbreviation of the query through ComBoxFactory at the some places
	 
	
	protected boolean messageboardIsActive;
	
	
	// volatile variables .............

	private MessageBoard mb;
	
	boolean restarting=false;
	boolean denyAnyRequest = false ;
	
	int isSending=0;
	
	boolean messageDispatching=false;
	
	boolean deliveryCompleted;
	
	Vector<Long> sConfirmation = new Vector<Long>();
	
	Vector<Integer> assignedClientsideListenPorts = new Vector<Integer> ();
	
	int syncedSendCounter=0;
	
	// helper objects .................
 
	PrintLog out = new PrintLog(2,true);
	NumUtilities numutil = new NumUtilities();
	StringsUtil strgutil = new StringsUtil();
	ArrUtilities arrutil = new ArrUtilities (); 
	DFutils fileutil = new DFutils(); 
	
	// . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .
	public MessageBoard( String configpath , MessageBoardShareIntf msgboardshares ,ComBoxFactory comboxfactory, ClientMatcherIntf clientmatcher){
		
		String str="" ;
		
		 
		
		// int v = sendModeForBoard;
		comBoxFactory = comboxfactory ;
		configHomePath = configpath;
		
		msgBoardShares = msgboardshares;
		mbf = this.msgBoardShares.getMsgBoardFactory();
		
		 
		
		// the "msgBoardShares"-object guarantees, that these real global entities are shared 
		// between instances of class MessageBoard that are living within the same public MessageBoardInstance 
		msgBoardStorage  = msgBoardShares.getMsgBoardStorage();
		globalReceipts   = msgBoardShares.getGlobalReceipts() ;
		metaRegistration = msgBoardShares.getMetaRegistration() ;
		
		/* 
		    the creation of all of those has been moved to the factory , and references are copied to the MBShare!!
		    instead (as done above) we just fetch the references from the MBShare object, which also provides
		    an interface to access its stored references
		globalReceipts = new GlobalReceipts(this); 
		
		msgBoardStorage = new MsgBoardDataStorage( this, out ) ; OK
		
		msgBoardStorage.loadMsgBoardData() ;
		
		metaRegistration = new MetaRegistration(globalReceipts, msgBoardStorage);
		  
		*/
		 
		connectionsMgr = new ConnectionsManager( this ); 
		registrationMgr = new RegistrationManager( this ); 
		  
		mb = this;
		
		clientMatcher = clientmatcher ;
		
		homePath = msgBoardStorage.getHomePath() ; 
		
		
		// allowing for the threads
		messageboardIsActive = true;
		
		// on construction, it also loads persistently saved notifications
		waitingNotifications = new Notifications(this);  
		
		ipAddressList = NicAddresses.get();
		ipAddressListStr =  arrutil.arr2text( arrutil.changeArrayStyle(ipAddressList, ""),";" );
 
		
		hostingData = msgBoardStorage.readHostingData("1"); 
		
		// TODO: all config of "comBoxFactory" should be done through ComBoxProperties
		comBoxFactory.setReceptorListenPort(msgPortforReceptors  ) ;
		comBoxFactory.setSourceListenPort( msgPortforSources ) ; 
		comBoxFactory.setRegistrationPort( msgPortforRegistration ) ;
		 
		
		comBoxFactory.setCtrlPort(7111) ;
		comBoxFactory.setObserver(this) ;
		comBoxFactory.setDescription("bSR") ;  // comBoxFactory@6e41b5 ...   comBoxFactory
		
		
		// the ComCox is an abstraction of particular transports, we access it through the interface 
		comBox = comBoxFactory.createComBox() ; 
		
		protocolID = getComBox().getFactory().getProtocolID() ; // ComBox@1364dcb

		
		// v = sendModeForBoard;
		
		if (comBox.getReceiver() == null){
			out.print(3,"MessageBoard can't be started, thus the application will stop.");
			exitOnPortFailure();
			 
		}

		//
		// the messages arrive elsewhere, but we need in the connectionsMgr ;
		// we import the message via interface ConnectionsMgrIntf, which transports
		// the method connectionCheckResponse 
		
		// the participants' responses belong to the family of registry messages
		// i.e. they consists of xml which contains "<register>"
		connectionsMgr = new ConnectionsManager( this );
		
		// actually, it knows about metaRegistration, so we would not need to send/set registeredInstances on an extra move... 
		connectionsMgr.setRegistrations( metaRegistration.getRegisteredInstances()); // Vector<XRegistration>
		connectionsMgr.setStandardPeriod( standardPeriodForConnectionCheck ) ;

		
		
		registerMessageInterpreter = new RegisterMessageInterpreter( this,out) ;
		sourceMessageInterpreter = new SourceMessageInterpreter(this, filters, out);
		receptorMessageInterpreter = new ReceptorMessageInterpreter(this, out); // filters, 
		boardMessageInterpreter = new BoardMessageInterpreter(this, filters,out);
		
		// starting the threads in their wrapper classes
		// these handlers work on the FiFo queues of messages from sources and receptors, respectively
		// like so:  msgboard.interpreteSourceMsg(msgboard.sourceMessages.get(0)); ... 
		//           ... until all messages have been interpreted
		registerHandler = new RegisterHandler( this );
		srcHandler = new SourceHandler( this ) ;
		cliHandler = new ClientHandler( this ) ;
		mbHandler  = new BoardMessageHandler( this )  ;
		inBuffHandler = new InBufferHandler( this )  ;  
		notifyHandler = new NotificationHandler(this) ;
		
		
		out.print(3,"Message handlers have been started.");
		
		
		// like so: defined in MessageBoardFactory
		// comBoxFactory.setDesiredInfraStructure(  ); //ComBoxFactory.DRS_SUDP ) ;
		// or so:
		// comBoxFactory.setDesiredInfraStructure( ComBoxFactory.DRS_RESTLET ) ; 
		
 
		activeContextHeaderDescriptions = new ActiveContextHeaderDescriptions( this) ;
		
		messageMatcher = new MessageMatcher( this, out);
		msgmatcherProcess = new MessageMatcherProcess( this, messageMatcher, out);
		int r=0;
		
		 
		boolean isHttpRest = (comboxfactory.getProtocolID()  == MessageBoardFactoryProperties.protocolHREST);
		// MessageBoardFactoryProperties.__MESSAGEBOARD
		// && (isHttpRest==false)
		
		if ((sendModeForBoard==1) ){  // && (isHttpRest==false)
			if (comBox.getReceiver() == null){
				r = createBoardReceiver("receptors") ; // ???
			} else {
				r=1;
			}
			r = r + createBoardReceiver("sources") ; 
			r = r + createBoardReceiver("registry") ;	
		}else{
			if (isHttpRest==false){
				createHttpServer(); 
				this.msgPortAllInOne = this.httpServerPort;
			}
		}
		
		if (r<0){
			exitOnPortFailure();
		}
		if (comBox.getFactory().getProtocolID()==7){
			
		}
		
		// 
		Registration registration = new Registration( mbf, "BOARD" );
		
		// we never (should) have two different MessageBoards running on the same machine with the same protocol
		// thus we use the protocolID as a part of the name 
		simpleName = simpleName + this.comBoxFactory.getProtocolID();
		
		mxReg = registration.loadRegistration( simpleName, true );
		if (mxReg==null){
			mxReg = new XRegistration();
			mxReg.init( msgBoardShares.getMsgBoardFactory(), null) ;
			messageBoardsIdentifier = GUID.randomvalue() ;
			mxReg.setSuffixUID( messageBoardsIdentifier ); // this we need as unique identifier in the network of MessageBoards
			mxReg.setRole("BOARD") ;
			
			mxReg.setNameOfApplicant( simpleName+"-board-"+messageBoardsIdentifier );
			mxReg.setInstanceIsUnique(true); 
			mxReg.setInstanceID( comBoxFactory.getProtocolID()) ;
			
			registration.saveRegistration( simpleName , mxReg, 0);
			messageBoardsIdentifier = mxReg.getNameOfApplicant() ;
		}else{
			
		}
		// 
		
		// start the relay dispatcher only if we need it due to request
		if (mbf.getFactoryProperties().getRelayingIsAllowed()){
			taskRelayDispatcherProcess = new TaskRelayDispatcherProcess(this);
		}

		
		monitor = mbf.getMsgBoardMonitor();
		monitorProcess = new MonitorProcess(this, monitor);
		
		if (mbf.getFactoryProperties().getProtocolId() >= MessageBoardFactoryProperties.protocolTCP){
			// this checks for left overs... 
			// comparing the activity state of RECEPTORS and the state flags of transactions...
			cleanupCollector = new TaskItemsCleanupCollector(this) ;
		}
		
		
		if (r==0){
			out.print(3,"Receivers ('servers') have been started...");
			out.print(3,"MessageBoard has been prepared...");
		}
 
		
		
		// broadcast to service channel (port 7111) that re-registering is necessary
		// the MessageBoard has been restarted
		//

		str=str+"" ;
	}
	
	private void exitOnPortFailure(){
		System.err.println("\nNot all of the requested ports are available, thus the program will exit (5)");
		System.exit(5) ;
	}
	

	private String standardizeCallbackMarker( String rawXmlMsg){
		String str, originatorID="";
	
		if ((rawXmlMsg.trim().startsWith("<origin="))) { // &&(rawXmlMsg.contains("<register>")
			str = "";
			int p = rawXmlMsg.indexOf(">::");
			if (p > 0) {
				str = rawXmlMsg.substring(0, p);
				str = str.replace("<origin=", "");
				int p2 = str.lastIndexOf(".");
				str = str.substring(p2 + 1, str.length());
				originatorID = str;
				rawXmlMsg = rawXmlMsg.substring(p + 3, rawXmlMsg.length());
				// outx out.print(4, "Originator detected (2): " + str);
				
				str = "<originator id=\"" + originatorID + "\"/>";
				
				rawXmlMsg = rawXmlMsg.replace("</message", str + "\n</message");
				// e.g. str = "org.NooLab.glue.net.http.rest.resources.RegistrationsResource$MessageSupply@17d5d2a"
				//   -> str = "RegistrationsResource$MessageSupply@17d5d2a"
			}
		}
		return rawXmlMsg;
	}

	/**
	 * note that almost all messages are in xml format, with a certain pragmatics!
	 * (except when a connection is in streaming mode)
	 * 
	 * any incoming message is first checked here by the dispatcher, 
	 * before any parsing;
	 *  
	 */
	@SuppressWarnings("unused")
	protected void messageDispatcher( String rawXmlMsg){
		
		RequestHandlingWrapper requesthandlingwrap;
		boolean isByteStream, isSource, isMsgBoard, isBoard, isRegistrant, 
			    isClientSubscription, isRoleAvailable,isReceptor;
		
		String str, originatorID="";
		
		messageDispatching = true;
		
		// rebuilding the originator indication... 
		rawXmlMsg = standardizeCallbackMarker(rawXmlMsg);
		
		
if ( (rawXmlMsg.contains("<>")) && ((rawXmlMsg.contains("123"))) ){
	out.printErr( 2,"board dispatcher ... \n "+rawXmlMsg+"\n---------------\n");
}		

if ((rawXmlMsg.contains("<request content x")) && (rawXmlMsg.contains("RECEPTOR")==false)){
	 out.printErr(4, "... message about <request ...\n"+rawXmlMsg);
} // need the tid as new entry to add to sConfirmation<>
		// the global identifier: does it belong to our mechanism?
		isMsgBoard = rawXmlMsg.indexOf("<messageboard ")>0;
		isRoleAvailable = rawXmlMsg.indexOf("role=\"")>0;
		
												out.print(5,"   ... MessageBoard - message dispatcher (1), filled threads : " + collectThreadStates());
												
		if ((isRoleAvailable==false) || (isMsgBoard==false) ){
			// nothing happens with this message ... it simply gets lost 
			out.print(3,"   ... MessageBoard - messageDispatcher interrupted!");
			messageDispatching = false;
			return;
		}
		
		// 
		
		if (messageboardIsActive==false){
	
			out.print(2,"   ... MessageBoard requested, but already set to inactive.");
			return;
		}
		
		// ..................
		
												
															out.print(5, "dispatching message...");	
		isSource = (rawXmlMsg.indexOf("role=\"SOURCE\"")>0); //  || (rawXmlMsg.indexOf("role=\"IBOARD\"")>0);
		isReceptor = rawXmlMsg.indexOf("role=\"RECEPTOR\"")>0;
		isRegistrant = (rawXmlMsg.indexOf("<register>")>0) && (rawXmlMsg.indexOf("</register>")>0) ;
		
		// is the role of sender (or recipient) a MessageBoard? 
		// i.e. are we about to handle an inter-board message
		// 
		isBoard = rawXmlMsg.indexOf("role=\"IBOARD\"")>0;
		isByteStream  = rawXmlMsg.indexOf("xml")==0;  
		
		
		
		if (isRegistrant==false){
			isRegistrant = (rawXmlMsg.indexOf("<unregister>")>0) ;
		}
		
		if (isRegistrant){
			
												
												if (rawXmlMsg.toLowerCase().contains("receptor")==false){
													// out.print(2,"   ... MessageBoard - message dispatcher (reg) ...");
												}
												
			enteringRegistry( rawXmlMsg );
			// this will also create a "RequestHandlingWrapper{}", 
			//  -> calls a method 
			//    -> simple feeds a context specific message queue
			//      -> taken by an ongoing FiFo (separating the messages)
			//       -> routed to a new adhoc process (own thread)
			return;
		}
	 
		if (isSource){
												out.print(5,"   ... MessageBoard - message dispatcher (src) ...");
												out.print(6,"   ... raw message\n"+ ""+rawXmlMsg);
	        // each new incoming message from a SOURCE opens a new thread												
			requesthandlingwrap = new RequestHandlingWrapper( this, 2, rawXmlMsg,"" );
			requesthandlingwrap.go();
			
			messageDispatching = false;
			
			if (msgmatcherProcess!=null){
				if (rawXmlMsg.contains("<context")){
					msgmatcherProcess.setMessageContent( rawXmlMsg ) ;
				}
			}

		}
		
		if (isReceptor){
												// out.print(4,"   ... MessageBoard - message dispatcher (rec) ...");
			
			requesthandlingwrap = new RequestHandlingWrapper( this, 3, rawXmlMsg,"" );
			requesthandlingwrap.go();
			messageDispatching = false;
		}
		
		if (isBoard){
												out.print(4,"   ... MessageBoard - message dispatcher (iboard) ...");
			requesthandlingwrap = new RequestHandlingWrapper( this, 4, rawXmlMsg,"" );
			requesthandlingwrap.go();
			messageDispatching = false;
		}
		
		// put it also to the subscription handling process
		
		// -> MessageMatcher, which is running in a separate thread, organized by MessageMatcherProcess
		//    as with other processes, we introduce it through a shared collection of FiFo style
		   
		// as it is said... it is ONLY about matching the messages here....
		// another location where matching happens is in  
		// ReceptorMessageInterpreter::getNextSuitableTransactionIndex() -> SelectingTransactionUponRequest{}
		if (msgmatcherProcess!=null){
		 
			if (rawXmlMsg.contains("<context")){
				// msgmatcherProcess.setMessageContent( rawXmlMsg ) ;
			}
		}
		 
		
	} // end update()


	/**
	 * regardless the transport protocol, data are delivered here;
	 * this requires to correctly install the Observer 
	 * when creating the receivers (basicCliReceiver, basicSrcReceiver ) 
	 * via the method "comBox.createReceiver()" !! 
	 * 
	 * 
	 */
	public void update( Observable obsobj, Object objdata) {
		// these are messages from receiver 
		String msg,str="" ;
		
		
		if (restarting){
			// return a message to the sender proposing retry after a few seconds
			return;
		}
							out.print(5,"MessageBoard is receiving...   ");
	
		// expected: obj data should be an XML ...
		msg = (String)objdata;	
		
if (msg.toLowerCase().contains("receptor")){
	// out.print(2, "\n\n... msg from receptor ...\n");
	// out.print(5,msg+"\n\n");
}
		
		if (denyAnyRequest){          
			// we filter (do not react upon) "<unregister>" applications by the participants, since we shut down anyway
			if ((msg.contains("<unregister>")==false) && (msg.contains("<system>")==false)){
				out.printErr(2,"MessageBoard is going to shut down -> any request will be denied!");
				out.print(2, "...denied message was: \n\n-" + msg) ;
			}
			return;
			
		}
if (msg.toLowerCase().contains("<relay>")){
	out.printErr(2, "interboard-message arrived in MessageBoard ... \n" +msg);
}
						    if (obsobj!=null){
						    	str = obsobj.toString()+"; ";
						    }
							out.print(5, "\n--------------\nMessageBoard (is active:"+messageboardIsActive+"): msg received : ");
							out.print(6 ,str+	"\n"+ msg+"\n--------------\n\n" );
							out.print(5,"   ... MessageBoard - message dispatcher (a), filled threads : " + collectThreadStates());
							
							str = "   ... MB active = "+messageboardIsActive+
									    " ,  inbuffer : thread = a:"+inBuffHandler.getThrd().isAlive()+"/i:"+inBuffHandler.getThrd().isInterrupted()+
									    ", running = "+inBuffHandler.isRunning+
									    ", working = "+inBuffHandler.isWorking ;
							out.print(6,"...raw str is...  "+str);
		
		
		inBuffer.add( msg );
		
		if (inBuffHandler.isRunning==false){
			inBuffHandler = new InBufferHandler( this )  ;
			delay(5);
			inBuffer.add( msg );
		}
		                    out.print(5,"   ... inbuffer now containing : "+ inBuffer.size()+" items.\n\n");
		
		// the inBuffer is just sth like the welcome desk, 
	    // from here (and its FiFo process), we proceed to "messageDispatcher()" (see above)		                   
		                    
		
		
		// the only thing we do here is to put the incoming stuff into a collection
		// in a (almost) non-colliding manner
		// just collecting the object as String is quite fast
		// "as String" is necessary to unlink the reference
		
		// this buffer acts as a FiFo, which is worked on in a dedicated thread object
		// there are additional, slower FiFo buffers for the various roles
		
		// if 2 messages really would collide, then the messages would be corrupted,
		// no receipt would be returned, and the originator of the message would retry
		
	}
	// . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .
	 
 
	
	class RegistryMsgInterpretationProcess extends MsgInterpretationProcessAbs {
		
		public RegistryMsgInterpretationProcess( Message msg, String role){
			super(msg, role+"MsgInterpretationProcess");
		}

		 
		protected void perform() {
			int r;
			// now we have separated registry request flowing in shortly after each other
			r = registerMessageInterpreter.interpretingSingleRegMessage(msg);
			
			
			if (r<0){
				
			}
		}
	}
	
	
	
	protected void interpreteRegistryMessages( Message msg ){
		// each message gets its own process, which then will return to the "perform()" just above
		new RegistryMsgInterpretationProcess( msg ,"REG").go(); 
	}

	/** 
	 * the purpose of this class is to singularize incoming messages into single threads 
	 * this class is created upon each new message, so, message handling does not overlap 
	 * 
	 */
	class SourceMsgInterpretationProcess extends MsgInterpretationProcessAbs {
 
		public SourceMsgInterpretationProcess( Message msg, String role){
			super(msg, role+"MsgInterpretationProcess");
		}
		
		protected void perform(){
			int r;
			// these objects are global inside the MessageBoard
			sourceMessageInterpreter.setTransactions(transactions) ;
			r = sourceMessageInterpreter.interpretingSingleSourceMessage(msg);
			
			if (r<0){
				out.printErr(3, "message from SOURCE could not be interpreted.");
				out.printErr(4, msg.rawXmlMsg+"\n-------------------------------\n");
			}
		}
	}
	
	protected void interpreteSourceMsg( Message msg){
		
		new SourceMsgInterpretationProcess(msg,"Source").go(); 
		
	}
	
	
	class BoardMsgInterpretationProcess extends MsgInterpretationProcessAbs {
		
		public BoardMsgInterpretationProcess( Message msg, String role){
			super(msg, role+"MsgInterpretationProcess");
		}
		
		protected void perform(){
			int r;
			// these objects are global inside the MessageBoard
			boardMessageInterpreter.setTransactions(transactions) ;
			r = boardMessageInterpreter.interpretingSingleBoardMessage(msg, mbf.getMsgBoardConnector() );
			
			if (r<0){
				// interpreting failed...
			}
		}
	}
	
	
	protected void interpreteInterBoardMsg( Message msg){
		
		new BoardMsgInterpretationProcess(msg,"Board").go(); 
		
	}	
	
	class ReceptorMsgInterpretationProcess extends MsgInterpretationProcessAbs {
		 
		public ReceptorMsgInterpretationProcess( Message msg, String role){
			super(msg, role+"MsgInterpretationProcess");
		}

		protected void perform() {
			receptorMessageInterpreter.setTransactions(transactions) ;
			receptorMessageInterpreter.interpretingSingleReceptorMessage(msg);
		}
	}
	
	protected void interpreteReceptorMsg( Message msg  ){
		new ReceptorMsgInterpretationProcess(msg,"Receptor").go();
	}
 

 
	/**
	 * 
	 * is is ONLY to be used for result notifications
	 * 
	 *
	 */
	class NotificationHandlingProcess implements Runnable{
		
		long currentTime;
		Notification noty ;
		Thread notyThrd ;
		
		public NotificationHandlingProcess(long currentTime, Notification noty){
			this.noty = noty;
			this.currentTime = currentTime;
			
			notyThrd = new Thread(this,"NotificationHandlingProcess")  ;
		}
		
		public void go(){
			notyThrd.start();
			
		}
		
		
		public void run(){
			
			handlingSingleNotificationAboutResults();
			
			try{
				out.print(6, "re-joining NotificationHandlingProcess");
				// notyThrd.join() ;
				
			}catch(Exception e){ }
			
		}
	
		private boolean prelimsFulfilled( Notification note){
			
			boolean prelimsFulfilled;

		    prelimsFulfilled = true;
		    
		    		    
			if (note == null) {
				prelimsFulfilled = false;
			}
			if ((note.getRole().length() == 0) || (note.getFilename().length() == 0) ) {
												out.printErr(2,"\nMessageBoard: handling notification : buffer not found");
				prelimsFulfilled = false;
			}

			if ((protocolID < MessageBoardFactoryProperties.protocolTCP)){
				 prelimsFulfilled = false; 
			}
			return prelimsFulfilled;
		}
		
		private int handlingSingleNotificationAboutResults() {
					
				 
					long transID, instanceID;
					int resultsReturnPort=-1,k1,k2,ix,sv, poll = 0;
					boolean  timeOk, tisAligned ;
					String roleOfIssuer;
					Notification note;
					XRegistration xreg = null;
					
					Transaction transaction=null;
		
					
					if (transactions == null){
						return -1;
					}
					if (noty==null){
						return -1;
					}
					
					poll=9;							
				    note = noty;
				    			// call the treatment on the particular notifications this treatment decides, whether to send, 
					// and also it removes items from the list (if successful... if not, it will again be added to the end of the list)
				
					// check interval , not too frequent... time check is per notification
					// lastCheckingTime = note.getLastEventTime();
					// timeOk = (System.currentTimeMillis() - lastCheckingTime) > 5000;
					timeOk = true; // already checked in the basic loop of thread "NotificationHandler{}"
					transID = note.getResultsOfTransactionID();
					
					ix = transactions.getIndexByTransactionID(transID,1);
					transaction = transactions.getItem(ix) ;
					sv = transaction.getTransactState() ;
					
					if (sv >= Transaction.TS_RESULTS_RETURNED){
						 
						
						
						if (note!=null){
							// out.print(2,"\nMessageBoard: handling notification, transaction (tid:"+note.getNotificationID()+") already returned ?");
						}
						
						note=null;
						return -3;
						
					}else{
						
					}
					
					transID = note.getResultsOfTransactionID();
												
					if ((timeOk) && (prelimsFulfilled(note))){
														// this out-message only if the notification is old... (should be monitored for its age) 
														out.printErr(4,"\nMessageBoard: notifying SOURCE about delayed transaction (tid:"+transID+")... ");
														out.print(2,"\nMessageBoard: handling notification about results (note-id:"+noty.getNotificationID()+", tid:"+transID+")");
		// ATTENTION notification ID is NOT necessarily == transaction ID !! there is a dedicated field in the Notification for the tid
						poll=0;
						
						if (mb.protocolID >= MessageBoardFactoryProperties.protocolTCP){
		
							 
							
							instanceID = transaction.getInstanceID();
		
							// // return true even if alignment is not requested
							tisAligned = transactionIsAligned( instanceID,  transID);
		 
							if (tisAligned){ 
								// if it is aligned then we may send it immediately, 
								// yet, the NotificationHandler process also runs for TCP ! 
								xreg = XRegistration.getItemByID( instanceID, mb.metaRegistration.getRegisteredInstances()) ;
								
								// 1. indicate to stop all notifications, any notification process must be completed first
								//
								notificationIdlingValue=1;   
								// 2. then wait for this completion, note that the handled notification ( get(0)) is
								//    temporarily removed from the list of notifications!
								// 	  hence, simply waiting is ok
								//    we wait until the NotificationHandler loop sets the idle indicator to =2 
								// while ( notificationIdlingValue==1){ delay(1);}
								// no need for this we are INSIDE the NotificationHandler Process
								
								waitingNotifications.removeNotificationByTransactionID(transID) ;
								
								// 3. sending 
								handlingSafelySendingResults( transaction, note );
								
								// 4. removing, if it is still there
								
								removeTaskToAlignment(instanceID,  transID);
								
								// 5. if alignment is NOT requested we switch the NotificationHandler ON again...
								if ( (xreg!=null) && (xreg.getAlignedResultsBasketSize()<=0)){
									notificationIdlingValue=0;
								}
								
								// if alignment is requested we switch to a dedicated handler, which 
								// stops all notifications for this instanceID, 
								// after disabling the notifications for this instance, we will
								// switch the handler ON for the rest again
								
								if ( (xreg!=null) && (xreg.getAlignedResultsBasketSize()>=1)){
									
									caringForWaitingAlignedItems( xreg, instanceID );
								}
								
								notificationIdlingValue=0;
								
								return 0;
							}  
							
						}
						
						// create notification message
						
		
						// either we save transaction too, or we put the instanceID into the note
						if (transactions != null){
							transaction = transactions.getTransactionByID(transID);
							// the result object in the transaction should be filled with b64 encoded object string!!
							// 
						}
						
						if (transaction==null){
							// MANDATORY: the transaction MUST exist, either online, or via re-loading!!
							out.printErr(3,"\nMessageBoard: handling notification : transaction not found, requested id="+transID+"; not notification issued.");
							// TODO: we have to close the transaction (set errcode!) and remove the bufferfile
							instanceID = -3;
							resultsReturnPort=0;
							// if not, we prevent its reloading (it has been removed by the NotificationHandler loop)
							poll=0 ;
							return 0;
						}else{
							// this works only, if the instance is persistent,
							// if it would not be persistent, then the instanceID would not match;
							// here, in the context of the MessageBoard, it should be always possible to find the transaction
							// 
							instanceID = transaction.getInstanceID();
							
						}
		
						roleOfIssuer = note.getRole();
		
						if (roleOfIssuer.length() == 0) {
							roleOfIssuer = "RECEPTOR";
						}
		
						// sending data
						if (instanceID>0){
							xreg = retrieveRegistrationObject(instanceID);
		
							if (xreg == null) {
								out.printErr(4,"\nMessageBoard: handling notification : registration not found.");
								poll = 7;
							}
						}
						
						
						/*
						 * creating a small XML which offers the SOURCE participant to fetch this results package
						 */
						if ( (poll==0) && (xreg!=null) && (xreg.isInstanceConnectionConfirmed()) && (xreg.getConnectionState()>=1)){ // the participant might have disappeared...
							
							
							k1 = waitingNotifications.size();
							 
							
							tisAligned = transactionIsAligned( xreg.getInstanceID(),  transID); //tiab
		
							if (tisAligned){
								
								int r = deliverResultsNotification( note, roleOfIssuer, transID, xreg );
								
								if (r==0){
									// of course we can not close it right it, we first have to deliver it upon 
									// getting contacted by the client (usually the SOURCE here)
									
									// Q out.print(2, "delivery of a notification about available results from transaction id="+transID+" has been successful, so it will be closed");
									// Q transaction = transactions.getTransactionByID(transID) ;
									// Q transaction.setTransactState(Transaction.TS_CLOSED) ;
									transactions.save() ;
									waitingNotifications.removeNotificationByTransactionID(transID) ;
								}else{
									if (protocolID >= MessageBoardFactoryProperties.protocolTCP){
										out.printErr(3, "delivery of a notification about available results somehow failed (err:"+r+")");
									}
								}
									
							}
							k2 = waitingNotifications.size();
						}
						
						// 
						if (resultsReturnPort<0){ 
							if (protocolID >= MessageBoardFactoryProperties.protocolTCP){
								out.printErr(4, "Unknown port, or no information avilable about participants port for delivering results."  );
								poll=3;
							}  
						}
					
						note.setLastEventTime( currentTime );
					} // timeOk ? 
				
					
					
					if (poll >= 1) {
						// in case of problems, we switch to polling for
						// this result snip
						
						note.setActivelyInforming(true); // if false, then the client (SOURCE) has to ask about available notifications
						msgBoardStorage.saveNotificationsDictionary(waitingNotifications);
		
						// reintroduce this notification again into the list of waiting notifications
						if (note!=null){
							k1 = waitingNotifications.size();
							
							waitingNotifications.add( note );
		
							k2 = waitingNotifications.size();
							waitingNotifications.save();
							k1 = k2-k1;
						}
					}
					return 0;
				
				}

		/**
		 * if alignment is requested we switch to a dedicated handler, which 
		 * stops all notifications for this instanceID, 
		 * after disabling the notifications for this instance, we will
		 * switch the handler ON for the rest again
		 * 
		 * note that the notification is "by-passed" globally for ALL instances while we are
		 * marking the notes for this instance !!
		 * 
		 * @param xreg
		 * @param instanceID
		 */
		private void caringForWaitingAlignedItems( XRegistration xreg, long instanceID ){
			
			long iid,transID;
			int ix;
			boolean tisAligned;
			Transaction transaction;
			Notification note;
			
			// removeTaskToAlignment(instanceID,  transID);
			
			 
			// are there others that could be returned ?
			if ( (xreg!=null) && (xreg.getAlignedResultsBasketSize()>=1)){
								String str = strgutil.arr2text(  xreg.getAlignmentQueue(), ";");
								if (xreg.getAlignmentQueue().size()>0){
									out.printErr(3, "--- waiting items : "+str);
								}
						
				// preventing the notifications for this instanceID by setting a particular flag
				for (int i=0;i<waitingNotifications.size();i++){
					
					note = waitingNotifications.get(i) ;
					iid = note.getIssuerInstanceID() ;
					
					if (iid == instanceID){
						note.setByPassProcess( true );
					}
				}
					
				// switching by-pass of NotificationHandler OFF -> Notifications will be issued again
				// we by-passed it only for a few millis
				notificationIdlingValue = 0; 			
								
				int q=0;					
				while (q<waitingNotifications.size()){
					
					note = waitingNotifications.get(q) ;
					
					transID = note.getResultsOfTransactionID() ;
					
					tisAligned = transactionIsAligned( instanceID,  transID);
					
					if (tisAligned ){
						// select note by transaction ID
						transaction =null;
						// note = waitingNotifications.getNotificationByTransactionID(transID) ;
						ix = transactions.getIndexByTransactionID(transID,1);
					
						if ((ix>=0) && (note!=null)){
							transaction = transactions.getItem(ix) ;
							instanceID =transaction.getInstanceID() ;
							
							waitingNotifications.removeNotificationByTransactionID(transID);
							
							handlingSafelySendingResults( transaction, note );
							
							removeTaskToAlignment( instanceID,  transID);
							// 
							
							
						}
						delay(50);
						
					}else{
						// proceed only if the first one was aligned, else leave... and wait for the next match
						break;
					}
					q++;
				}
				
				// making the notifications for this instanceID available again by re-setting the flag
				for (int i=0;i<waitingNotifications.size();i++){
					
					note = waitingNotifications.get(i) ;
					iid = note.getIssuerInstanceID() ;
					
					if (iid == instanceID){
						note.setByPassProcess( false );
					}
				}

			}
			
		}

		// this is the alternative for handlingSingleNotificationAboutResults() in case of direct protocols !
		private void handlingSafelySendingResults( Transaction transaction, Notification note) {
		
			String xmlMsg="", bufferfilename, str;
			int resultsReturnPort,r ;
			long tid, iid;
			XRegistration xreg = null;
			Transaction transact;
			XMessage xmsg = new XMessage(out);
			// .................................................
			
			// we should checked that already, anyway...
			if (mb.protocolID < MessageBoardFactoryProperties.protocolTCP){
				return ;
			}
			if ((transaction==null) || (note==null)){
				return;
			}
			
			tid = transaction.getTransactionID();
			iid = transaction.getInstanceID() ;
			
			if (tid>0){
				xreg = retrieveRegistrationObject(iid);
			}	else{
				return;
			}
			if (xreg==null){
				return;
			}
			
			resultsReturnPort = xreg.getClientListeningPort() ;
			if (resultsReturnPort <0){
				return;
			}
			
			// .................................................
			// loading the buffer, but we do not need the transaction, instead we need the result buffer
			 
			bufferfilename = msgBoardStorage.createTransactionBufferfileName( iid,  tid, "BOARD", "BOARD") ;
							 if (fileutil.fileexists(bufferfilename)==false){
								 out.print(2, "While preparing sending the result package, the bufferfilename was not found...");
								 out.print(2, "expected file would have been : "+bufferfilename);
								 return;
							 }
			try {
				
				Object obj = msgBoardStorage.readTransactionDataObj(bufferfilename);
				// xmlMsg = (String)obj ;
				transact = (Transaction)obj;
				
				xmlMsg = xmsg.createReturnOfDataUponNotification( transaction, "BOARD", transaction.getClientSecret() );
				
			} catch (Exception e) {
			}
			// sending... yet for sending we should use an independent process, the deliveryPad or the Regsender...
			
			
			
			str = (String)transaction.getDataObject();
			
			
			ResultSender resultSender = new ResultSender( resultsReturnPort, xmlMsg, tid, "result sending process");
		
			resultSender.periodLength = 3000; // millis between trials
			resultSender.count = 3; // max trials for this sender's instance
		
									out.print(3, "waiting ("+(resultSender.periodLength/1000)+") for confirmation of transaction "+tid+" started..."); 
									out.print(4, "XML result package is...  \n"+ xmlMsg);		
		
			r = resultSender.waitForDelivery();
			// the return has to set an entry in  sConfirmation.indexOf(transid) !!
			/*
			    if successful then...
			    
			 	transaction = transactions.getTransactionByID(transID) ;
				transaction.setTransactState(Transaction.TS_CLOSED) ;
				transactions.save() ;
				waitingNotifications.removeNotificationByTransactionID(transID) ;
		
			 */
			
		}

		private int deliverResultsNotification( Notification note, String roleOfIssuer, long transID, XRegistration xreg ){
			XMessage xmsg = new XMessage(out);
			String xstr,secret;
			int resultsReturnPort, poll,r=-1 ;
			long notificationID;
			ResultSender resultSender;
			
			
			notificationID = note.getNotificationID() ;
			
			secret = GUID.randomvalue() ;
			xstr = xmsg.createResultsNotificationMsg("BOARD", roleOfIssuer, secret, notificationID, transID, xreg.getInstanceID());
		
			resultsReturnPort = xreg.getClientListeningPort() ;
		
			if (resultsReturnPort>0){
				try {
					if (note.getPolling() < 0) {
						// it it is not in the queue of already touched notifications 
						// we try to return it actively
						poll = -4;
						note.setActivelyInforming(true);
					}
		
					if (note.isActivelyInforming() == true) {
						poll = -2;
						
						note.setNotificationProcessRunning(true);
						
						// we ALWAYS use triggered poll, because we can not be sure whether the client is actually available !
						// if not, we have to buffer the results ... => so we do buffering ALWAYS
						out.print(3, "installing sender (towards port: "+resultsReturnPort+") for triggering poll by source...");
		
						// we need a small thread for that (like in registration), otherwise the
						// notification handler will stop ??
						// the confirmation for receiving the results by the SOURCE will be
						// routed to "interpreteSourceMsg()", and handled there
		
						
												out.print(2, "notification message is:  \n"+ xstr);
						
						// this class wraps: mbSender.send(xstr);
						resultSender = new ResultSender( resultsReturnPort, xstr, transID, "notification process");
		
						resultSender.periodLength = 11000; // millis between trials
						resultSender.count = 40; // max trials
		
												out.printErr(2, "waiting ("+(resultSender.periodLength/1000)+") for acceptance of notification for transaction "+transID+" started..."); 
												out.print(5, "notif.str is...  "+ xstr);		
		
						r = resultSender.waitForDelivery();
						
						
						
						if (r==0){
							
							waitingNotifications.save();
							poll=0;
						}else{
							
							if (r<=-1){
								 
							}
						}
						out.print(3, "...waiting for acceptance of notification for transaction "+transID+" finished, result: " + r);
						 
						note.setNotificationProcessRunning(false);
						// then, if successful, deliver xml to source: = xMessage.createDataDeliveryMessageFrame()
					} else {
						poll = 1;
					}
		
				} catch (Exception e) {
					note.setPolling(2);
					poll = 2;
				}
		
				
			} // clientListeningPort is available
			
			return r;
		}

		class ResultSender implements Runnable{
			
			DataSenderIntf mbSender ; 
			
			String xml;
			boolean timedOut;
			Thread regsendingThrd;
			int timeOut = 7000, resultsReturnPort;
			int periodLength = 1000;

			int count = 2, sendProcessActivation=0;
			boolean enforcedStop = false ;
			long transid;
			String description;
			Timer timer;
			TTask ttask;
			

			public ResultSender( int port , String xml, long tid, String description) {

				resultsReturnPort = port;
				this.xml = xml;
				this.transid = tid;
				regsendingThrd = new Thread(this,"ResultSender-"+port);
				
				this.description = description;
				/*
				timer = new Timer(true);
				ttask = new TTask(description);
				timer.schedule(ttask, timeOut);
				*/
			}

			@SuppressWarnings("static-access")
			public void requestStop() {
				enforcedStop = true;
				try {
					Thread.currentThread().sleep(200);
				} catch (InterruptedException e) {
				}

				try {
					regsendingThrd.join();
				} catch (InterruptedException e) {
				}
			}


			private void regSendingHandler() {

				int z,r ;
				DelayFor delayFor = new DelayFor() ;
				
				mbSender = comBox.createSender( resultsReturnPort, 7111,"mbrr") ;
				
				out.print(3, "... thread ResultSender, regSendingHandler(): sending request : " + description);

				deliveryCompleted = false;
				enforcedStop = false;
				z = 0;
				count = 5 ;
				
				
				// SenderWrap sender = new SenderWrap(messageBoard);
				// sender.sendingMessage(xmloutmsg, deliveryPort);
				
				SenderWrap sender = new SenderWrap(mb);
				if (periodLength<50){
					periodLength=50;
				}
				
				while ((z < count) && (enforcedStop == false) && (deliveryCompleted == false)) {

												out.print(4, "... just before <sender.sendingMessag()>, going to send  to participant: " + description + ", delivery completed: " + deliveryCompleted+" ,  sender object : "+ this.toString());
												out.print(4, "... protocol to be used by ComBox : "+ getComBox().getFactory().getProtocolID());
												if (getComBox().getFactory().getProtocolID()>= MessageBoardFactoryProperties.protocolTCP ){
													out.printErr(4, "... targeted remote port of participant : "+resultsReturnPort);
												}
					// mbSender.send(xml);
					r = sender.sendingMessage( xml, resultsReturnPort) ;
						if (r<0){
							out.printErr(3, "sendingMessage() failed in NotificationHandlingProcess:ResultSender::regSendingHandler(), code="+r);
						}
					try {
						// stepwise waiting... does not make much sense here...
						// so far we do not have an indicator for flexible control...
						// delayFor.period(periodLength, 100);
						out.delay(periodLength); // e.g. 3000

					} catch (Exception e) {
					}
					z++;
				}
				sender=null;
				//delayFor.setUserbreak(true);
				out.print(4, "... sending to participant: " + description + ", delivery INTERRUPTED ,  sender object : "+ this.toString());
				
			}

			@Override
			public void run() {
				regSendingHandler();
			}

			@SuppressWarnings("static-access")
			public int waitForDelivery() {
				
				int z,p, result = -1;
				int limit = 12; // if debug -> ">1000"
				z = 1;
				enforcedStop = false;
				regsendingThrd.start();
				
				// "processCompleted" is set in update() =>  handlingRegistrationReply()
				while ((deliveryCompleted == false) && (timedOut == false) && (enforcedStop==false)) {

					deliveryCompleted = (sConfirmation.indexOf(transid)>=0 );

					if (sConfirmation != null) {
						// out.print(3, "");
					}
					timedOut = (z > limit); // 100 = approx. 10 seconds
					if ((z > 2) && (z % 50 == 0)) {
						if ((z < 55) && (z > 5)) {
							// out.print(3,"");
						}

						out.print(5, "obj sConfirmation = " + sConfirmation);
						// System.out.print(".");

					}
					;
					try {
						if (deliveryCompleted == false) {
							mb.delay(20);
						}

					} catch (Exception e) {
					}
					z++;
				} // ->
				
				if (deliveryCompleted == true) {
					result=0;
					p = sConfirmation.indexOf(transid);
					while (p>=0){
						if (p>=0){
							sConfirmation.remove(p) ;
						}
						p = sConfirmation.indexOf(transid);
					}
				}
				
				return result;
			}

			class TTask extends TimerTask {
				String description;

				public TTask(String description) {
					this.description = description;
				}

				public void run() {

					out.print(4, "Requesting stop of " + description + "...");
					requestStop();
				}

			}

			public void setPeriodLength(int periodLength) {
				this.periodLength = periodLength;
			}
		} // inner inner class ResultSender

	} // inner class 
	
	/**
	 * we are coming from the NotificationsHandler, which works on the collection "waitingNotifications", 
	 * sending its items one by one to here
	 * 
	 * the list of "waiting " is created in  "interpreteReceptorMsg(xml)"
	 *   
	 * 
	 * "Notifications" are always about results.
	 * By default results are saved to a filename, which allows to handle
	 * polling and pushing in a similar manner : we always read it from file
	 * 
	 * 
	 *
	 * 
	 */
	public void handlingNotification( long currentTime, Notification noty ){
		
		try{
			(new NotificationHandlingProcess(currentTime,noty)).go();
			// this creates a thread and calls its own worker method "handlingSingleNotificationAboutResults()"
			
		}catch(Exception e){
			e.printStackTrace() ;
		}
	}


	
	
	protected void handlingRequestForRegistering( String rawXmlMsg, String thrdGUID ){
			
			int k=0;
			
		    if (srcHandler.isRunning() == false){
		        srcHandler.startHandler();
		    }
		       
			Message msg = new Message(); 
			msg.setMessageboard(this);
			
			msg.setMsg( rawXmlMsg, registerMessages.size() );
			msg.setGUID(thrdGUID) ;
			
			// put it to the buffer;
			registerMessages.add( msg ) ;
			k = registerMessages.size();
			 	                       out.print(6,"\nmessage from source has been arriving at the message board...   ");
			 	                       out.print(5,"message buffer for sources now contains "+k+" messages.   ");
			 	                              				
		    k=k+0; 
	}
	
	
	protected void handlingMsgFromReceptor( String rawXmlMsg ){
		 
		Message msg = new Message(); 
		msg.setMessageboard(this);
		
		// put it to the buffer;
		msg.setMsg( rawXmlMsg, receptorMessages.size() );
		receptorMessages.add( msg ) ;
		
	}
 
	protected void handlingMsgFromBoard( String rawXmlMsg ){
		 
		Message msg = new Message(); 
		msg.setMessageboard(this);
		
		// put it to the buffer;
		msg.setMsg( rawXmlMsg, boardMessages.size() );
		boardMessages.add( msg ) ;
		
	}
	
	protected void handlingMsgFromSource( String rawXmlMsg ){
		/*  variant 1:
			<messageboard name="spela" role="SOURCE">
				<request type="transactionID">
	  				<role value="SOURCE"/>
	  				<instance id="234"/>
				</request>
			</messageboard>
			===========================================
			variant 2:
			<messageboard name="spela" role="SOURCE">
				<request type="delivery">
	  				<role value="SOURCE"/>
	  				<transactionID value=""/>
	  				<instance id="234"/>
				</request>
			</messageboard>
			
		*/
		
		int k=0;
		
	    if (srcHandler.isRunning() == false){
	        srcHandler.startHandler();
	    }
	       
		Message msg = new Message( ); 
		msg.setMessageboard(this);
		
		// put it to the buffer;
		msg.setMsg( rawXmlMsg, sourceMessages.size() );
		sourceMessages.add( msg ) ;
		k = sourceMessages.size();
		 	                       out.print(6,"\nmessage from source has been arriving at the message board...   ");
		 	                       out.print(5,"message buffer for sources now contains "+k+" messages.   ");
		 	                              				
	    k=k+0; 
	}
 
	public void handlingInternalControlTask( long instanceID, String originatorID, String taskLabel, String parameters, String rawXmlMsg ){
		
		new ControlTaskHandler(this, originatorID, instanceID, taskLabel, parameters ,rawXmlMsg, out ).go();
	}

	/**
	 * 
	 * these disconnects send a message to the client such that the client will initiate the disconnect<br/>
	 * 
	 * in contrast, the method (below) ""unregister()" does not rely on the client's activity.
	 * 
	 * 
	 */
	private void disconnectRegisteredInstances( int justInform){
		XRegistration xreg ;
		
		
		for (int i=0;i<metaRegistration.size();i++){ 
					// refers to registeredInstances
			xreg = metaRegistration.getItem(i) ;
			disconnectRegisteredInstance( xreg, -justInform ); // XRegistration
			
		}// i-> all registered instances
		
	}
	@SuppressWarnings("unused")
	private void disconnectRegisteredInstance( XRegistration xreg ){
		disconnectRegisteredInstance( xreg,0); 
	}
	
	private void disconnectRegisteredInstance( XRegistration xreg , int restart){
		  
		int port;
		String host ;
		
		XMessage xmsg ;
		xmsg = new XMessage(out) ;
			
			port = xreg.getClientListeningPort() ;
			host = xreg.getHostname() ;
			
			
			if (this.protocolID >= MessageBoardFactoryProperties.protocolTCP ){
				out.print(2, "sending shutdown message to instance "+xreg.getInstanceID()+" of type "+xreg.getRole()+", using port "+port+" @ "+host) ;
				sendShutDownMsg( xmsg, xreg, port, host,restart) ;
			}
			 
	}

	

	public synchronized void syncedSend( int targetPort, String xstr ){
		
		 
		DataSenderIntf mbSender ;
		
		syncedSendCounter++ ; //yyyyyyyyyyyyyyyyyyyyyyyyyyyyy
		
		//SenderWrap sender = new SenderWrap(this);
	
		mbSender = comBox.createSender(targetPort, 7111, "");
		// routes to constructors like new UdpDataSender( regConfirmPort, 7111);
	
		mbSender.setTimeOut(100);
							out.print(3, "   ... sending reg confirm to port : " + targetPort + " , sender object ("+syncedSendCounter+") : " + mbSender);
	
	
		// in case of httpc, this is NOT a physical send, instead, this "send" relays the
		// the message back to the http session object through switching observers
		mbSender.send(xstr);
			 
							out.print(4,"   ... reg confirm sent ("+syncedSendCounter+") to port : " + targetPort );
	}
	
	
	
	synchronized public int ensureUniqueListeningPortForParticipant( int cport){
		int port = cport ;
		boolean portAlreadyAssigned;
		
		if ( assignedClientsideListenPorts.contains(port)){
			portAlreadyAssigned = true;
			
			while ( portAlreadyAssigned ){
				port++;
				portAlreadyAssigned = (assignedClientsideListenPorts.contains(port)==true);
			}
			assignedClientsideListenPorts.add(port);
			
		}else{
			assignedClientsideListenPorts.add(port);
		}
		
		out.print(4, "server side port list:");
		for (int i=0;i<assignedClientsideListenPorts.size();i++){
			// System.out.print( assignedClientsideListenPorts.get(i)+", ") ;
		}
		 
		return port;
	}
	

	synchronized private void enteringRegistry( String rawXmlMsg ){
		RequestHandlingWrapper requesthandlingwrap;
		
		
		out.print(5,"   ... opening new handling thread ...");
		
		// starts a new thread and calls "handlingRequestForRegistering()"	
		
		requesthandlingwrap = new RequestHandlingWrapper( this, 1, rawXmlMsg , GUID.randomvalue());
		requesthandlingwrap.go();
	
		// we arrive here from a FiFo buffer list, then this creates a new thread for each new registration request 
		// from here, we proceed to handlingRequestForRegistering()
	
		messageDispatching = false;
	
	}
	
		
	@SuppressWarnings("unused")
	private void sendShutDownMsg( XMessage xmsg, XRegistration xreg, int port, String host){
		sendShutDownMsg( xmsg, xreg, port, host,0);
	}
	
	private void sendShutDownMsg( XMessage xmsg, XRegistration xreg, int port, String host, int restart){
		String xmloutmsg="";
		int r;
		
		xmloutmsg = xmsg.createBoardShutdownMsg( xreg,restart  );
		
		SenderWrap sender = new SenderWrap(this);
		r = sender.sendingMessage( xmloutmsg, port);
			if (r<0){
				out.printErr(3, "sendingMessage() failed in sendShutDownMsg(), code="+r);
			}
		sender=null;
	}
	
	public boolean checkForNotifyHandlerRunning(){
		
		boolean rB=notifyHandler.isRunning;
		
		if (rB==false){
			notifyHandler = new NotificationHandler(this);
		}
		return rB;
	}
	
	public boolean checkTaskStackLimit(){
		boolean rB=true;
		int nt,limit;
		
		try{
			
			limit = mbf.getFactoryProperties().getTransactionStackSizeLimit();
		
			// count transaction assigned, not returned
			nt = mbf.getTransactions().countActiveTransactions();
				                 
			
			if (nt>=limit){
				rB=false;
			}
			 
		}catch(Exception e){
			rB=false;
		}
		
		return rB;
	}
	

	
	public void announceStop(){
		denyAnyRequest = true;
	}

	@SuppressWarnings("static-access")
	public void stopMessageBoard( boolean clientShutDown ){
		int z;
		boolean allStopped = false;

		connectionsMgr.stop();
		
		announceStop(); // --> denyAnyRequest = true;
		
		msgmatcherProcess.stopMMProcess();
		
		
		// put the message to the Source-, Client- handlers 
		// stopping threads
		messageboardIsActive = false; // this is a poison pill in the loop condition of the handler threads
		
		if (clientShutDown){
			disconnectRegisteredInstances(0); 
		
			// stop receivers
			comBoxFactory.stopAll();
		
			try {

				// waiting for all being stopped, and wait a max number of cycles

				z = 0;
				while ((allStopped == false) && (z < 510)) { // ==5 seconds

					Thread.currentThread().sleep(10);
					allStopped = collectThreadStates() <= 0;
					z++;
				}

				if ((z > 500) && (allStopped == false)) {
					srcHandler.enforcedStop();
					cliHandler.enforcedStop();
					// mbHandler.enforcedStop();
					inBuffHandler.enforcedStop();
					Thread.currentThread().sleep(200);
				}

			} catch (Exception e) {
				e.printStackTrace();
			}
		} else {
			// send a notification, that this MessageBoard will disappear
			if (this.protocolID >= MessageBoardFactoryProperties.protocolTCP){
				disconnectRegisteredInstances(1); 
				 
			}
		}
		// saving current data, and state of MessageBoard
		
		msgBoardStorage.saveMsgBoardData();
		
		waitingNotifications.save();
		transactions.save();
		// releasing all thread objects
		
		
		
		
		System.gc();
		
	}

	public int createHttpServer(){
		
		// create a restlet server
		
		// MessageBoardRestletServer
		// this creates a server component, based on the settings of the ComBoxFactory
		
		// comBoxFactory. ...
		uniformHttpReceiver = comBox.createReceiver( httpServerPort , 7111, this, "cSR");
		
		return 0;
	}
	
	public int createBoardReceiver( String domain){
		int result =0 ;
		if (domain.toLowerCase().contains("receptor")){
			basicCliReceiver = comBox.createReceiver(msgPortforReceptors , 7111, this, "cSR");
			if (basicCliReceiver==null){
				result = -1;
			}
		}
	
		if (domain.toLowerCase().contains("source")){
			basicCliReceiver = comBox.createReceiver(msgPortforSources , 7111, this, "cSR");
			if (basicCliReceiver==null){
				result = -1;
			}		
		} 
		if (domain.toLowerCase().contains("registry")){
			basicRegistrar = comBox.createReceiver( msgPortforRegistration , 7111, this, "bReg" );
			if (basicRegistrar==null){
				result = -1;
			}		
		}	
		
		return result ;
	}
	
	
	
	public XRegistration retrieveRegistrationObject( long instID ){
		
		XRegistration  xReg = null, xr;
		
		if (instID>0){
			for (int i = 0; i < metaRegistration.size(); i++) {

				xr = metaRegistration.getItem(i);
				if (xr.getInstanceID() == instID) {
					xReg = xr;
					break;
				}

			} // all registered instances
		}
		
		return xReg;
	}
	
	public int getReceptorCount( int stateValueLo, int stateValueHi, int mode){
		
		return getReceptorCount( stateValueLo, stateValueHi, "",mode);
	}
		
		
	public int getReceptorCount( int stateValueLo, int stateValueHi, String docType, int mode){
		
		int cc=0, sv;
		boolean hb;
		String role,xdocs;
		Vector<XRegistration> registeredInstances ;
		XRegistration xreg;
		
		
		registeredInstances = getRegisteredInstances();
		
		for (int i=0;i<registeredInstances.size();i++){
			
			xreg = registeredInstances.get(i) ;
			
			sv = xreg.getReceptorState() ;
			hb = false;
			
			role = xreg.getRole().toLowerCase() ;
				
			if (role.contains("receptor")){
				
				
				if (mode<0){
					hb = sv<=stateValueLo;
				}
				if (mode==0){
					hb = sv>=stateValueHi;
				}
				if (mode>=1){
					hb = (sv>=stateValueLo) && (sv<=stateValueHi);
				}
				
				if (hb){
					
					// is the docType matching ?
					xdocs = xreg.getAcceptType();
					
					hb = XRegistration.docTypeMatches( xdocs,docType );
					
					if (hb){
						cc++;
					}
				}
				
			} // is it a receptor at all ?
			
		}
		
		return cc;
	}
	
	public int getReceptorCount( int stateValue, int mode){
		/**
		 * TODO: we measure the time since the last census, only if >x seconds, we recalculate it
		 * such, we avoid overload by a dense network...
		 * 
		 */
		int cc=0, sv;
		boolean hb;
		Vector<XRegistration> registeredInstances ;
		XRegistration xreg;
		
		
		registeredInstances = getRegisteredInstances();
		
		for (int i=0;i<registeredInstances.size();i++){
			
			xreg = registeredInstances.get(i) ;
			sv = xreg.getReceptorState() ;
			hb = false;
			
			if (mode<0){
				hb = sv<=stateValue;
			}
			if (mode==0){
				hb = sv>=stateValue;
			}
			if (mode>=1){
				hb = sv==stateValue;
			}
			
			if (hb){
				cc++;
			}
			
		}
		
		return cc;
	}
	
	public void performTaskDispatch(){
		int bn, recCount=0 ;
		long sinceTime , delytime;
		String secret;
		MessageBoardConnector msgBoardConnector;
		Transaction transaction;
		Vector<XRegistration> availableReceptors;
		
		 
		
		if (boardAddresses==null){
			boardAddresses = this.msgBoardShares.getBoardAddresses() ;
		}
		
		if (boardAddresses!=null){
			bn = boardAddresses.countByState( BoardAddress.__BA_KNOWNMAP,1 ); //  
			
			
			if ((bn>0) && (transactions!=null) && (transactions.size()>0) ){

				int[] stateValues = new int[]{ Transaction.TS_TASK_CONFIRMED ,Transaction.TS_TASK_DENIED, Transaction.TS_TASK_ACCEPTED };
				
				long[] tiids = transactions.getTransactionIDsByStateValues( stateValues, 0) ;
				
				if ((tiids!=null) && (tiids.length >= 0)){
												out.print(4,"performTaskDispatch(), checking for available receptors ...");
					// get the count of registered and available receptors
					recCount = getReceptorCount( XRegistration.RST_IS_IDLE, XRegistration.RST_IS_AVAIL,1);
					
					// is there any task that is old ? 
					// or do we have no receptor registered ?
					if (recCount>0){
						// we check for old transactions only if there are some receptors registered,
						// if there are no receptors registered, we need not to check for the age of transactions
					
						out.print(2,"performTaskDispatch(), checking for due tasks...");
						
						// we "remove" those entries from the tiids list, which are not old
						for (int i=0;i<tiids.length;i++){
							if (tiids[i]>0){
								transaction = transactions.getTransactionByID(tiids[i]);
								delytime = transaction.getTimeofDelivery();
								sinceTime = System.currentTimeMillis() - delytime;
								
								if (sinceTime < ((long)(double)(1.4 * avgReceptorsReCheckTimevalue))){
									tiids[i] = -1;
								}
							}
						}
					} // recCount>0)
					
					for (int i=0;i<tiids.length;i++){
						
						
						if (tiids[i]>0){
					
							
							transaction = transactions.getTransactionByID(tiids[i]);
							
							if (transaction!=null){
								// setState is synchronized
								// nevertheless: collision avoidance...
								if (transaction.getTransactState() < Transaction.TS_TASK_STARTDELIVERY){
									
									transaction.setState( Transaction.TS_TASK_STARTDELIVERY ) ;
									delay(5 + (int)Math.random()*15);
									if (transaction.getTransactState()== Transaction.TS_TASK_STARTDELIVERY){ // still ? it could be that another process has been faster...
									
										transaction.setState(Transaction.TS_TASK_DELIVERING); //
									
										out.print(2,"performTaskDispatch(), relaying selected transaction (tid:"+tiids[i]+") to remote MessageBoard...");
										
										// now we have ask the remote MessageBoard, whether it knows about an available receptor, 
										// which could serve the document type and the context, other matchings will be checked over there
										// if it is denied , we will get informed here about a deniedRelayAccept
										
										msgBoardConnector = mbf.getMsgBoardConnector();
										
										msgBoardConnector.startTransactionRelay( tiids[i] );
										
										// we only handle 1 transaction at a time... so we leave the circus...
										break;
									} // assumed availability
								} // not being touched (state < TS_TASK_STARTDELIVERY)
							} // transaction!=null
							
						} // any tiids found that we could relay ?
					} // i->
				} // any transactions waiting ?
			} // size > 0
		} // collection != null
		
	} // performTaskDispatch
	
	
	
	public void handlingBlockedTransactions( String rawXmlMsg ){
		
		String xmloutmsg="";
		long transactionID, instanceID, timeofdelivery, td,ageofexpiry;
		int r,deliveryPort, errstate;
		boolean dropMsg;
		String ctyp, errWorded ;
		
		XRegistration xreg ;
		Transaction transaction;
		XMessage xmsg = new XMessage(out);
		
		
		try{
			

			dropMsg = ( getDropDelayUnrelayedMessages()<=1) ;
			
			timeofdelivery = xmsg.getTimeDataValue(rawXmlMsg, "delivery", -1) ; 
			ageofexpiry    = xmsg.getTimeDataValue(rawXmlMsg, "expiry", -1) ;
		 
			if (ageofexpiry>1){
				td = System.currentTimeMillis() - timeofdelivery;
				
				if (td < ageofexpiry){
					dropMsg = false;
				}				
			}
					
			transactionID = xmsg.getTransactionID(rawXmlMsg) ;			
			
			transaction = getTransactions().getTransactionByID(transactionID) ;
			instanceID = transaction.getInstanceID() ;
			
			xreg = XRegistration.getItemByID(instanceID, getRegisteredInstances()) ;
			deliveryPort = xreg.getClientListeningPort();
			
			
			if (dropMsg ){
				out.print(2, "data delivery for transaction "+ transactionID +" has not been relayed to receptor...") ;
				// inform the source about it: whether stored and delayed, or whether dropped    
				
				ctyp = transaction.getContentType() ;
				errstate = transaction.getTransactErrorState() ;
				if (errstate == Transaction.TS_ERR_NOMATCH){
					errWorded = "No matching participant found for requested content type '"+ctyp+"'" ;
					
				} else {
					errWorded = "ErrorCode = "+errstate ;
				}
				
				xmloutmsg = xmsg.createStateUpdateNotificationForSource( transactionID, transaction.getClientSecret(), 
																		 Transaction.TS_TASK_DENIED , errWorded, "BOARD");
				// 
				
			}
			if (dropMsg==false){ // a xml arranged differently by the SOURCE 
				out.print(2, "data delivery for transaction "+ transactionID +" will be delayed for an unknown amount of time...") ;
				
				xmloutmsg = xmsg.createStateUpdateNotificationForSource( transactionID, transaction.getClientSecret(), 
						 												 Transaction.TS_TASK_DELAYED , "No matching participant found.", "BOARD");
			
			}
			
			SenderWrap sender = new SenderWrap( this );
			r = sender.sendingMessage( xmloutmsg, deliveryPort );
				if (r<0){
					out.printErr(3, "sendingMessage() failed in handlingBlockedTransactions(), code="+r);
				}
			sender=null;
			
			if (dropMsg==false){
				// TODO:
				// store it and try it a bit later again, creating an internal message
				// handlingDelayedOfTransaction( transactionID, -1 ) ;
				// this stores the whole transaction as a binary, creates a "message" using the 
				// filename, puts that into a queue (which is run by a separate thread)
				// the thread checks if any RECEPTOR has announced its availability
				
				// handlingDelayedOfTransaction(transactionID);
				// and/or handlingReRoutingOfTransaction
				
				// sourceVariant_DataTaskQueueing( xmsg, rawXmlMsg ,rawIndicators);
				
			}
			
			
		}catch(Exception e){
			e.printStackTrace();
		}
		
	}
 
	/**
	 * this is a callback from ConnectionsManager's ConnectionChecker
	 * 
	 * 
	 * @param iid
	 */
	synchronized public void handlingLostConnection( long iid ){
		XRegistration xreg;
		int p;
		long xid;
		
		out.printErr(2, "Connection unexpectedly not available any more , iid:"+ iid);
		 
		for (int i=0;i<metaRegistration.size();i++){
			
			xid = metaRegistration.getItem(i).getInstanceID() ;
			xreg = metaRegistration.getItem(i) ;
			
			if (xid == iid){

				out.printErr(2, "removing instance "+ iid+" from collection of registered instances...");
			
				p = metaRegistration.getRegisteredInstances().indexOf(xreg) ;
				
				disconnectRegisteredInstance( xreg, 0 ); // 
				// 1==restart, 0 == shutdown : for keeping consistency, we must shutdown... 
				
				if (p<metaRegistration.size()){
					if (p<metaRegistration.size()){
						metaRegistration.removeItem(p) ;
					}
					
					int ks = metaRegistration.countSources() ;
					int nr = metaRegistration.countReceptors() ;
					out.print(1, "Instance removed, now there are "+ks+" sources and "+nr+" receptors.") ;
				}
				
				break;
			}
		}
	}
	
	/**
	 * see (above): disconnectRegisteredInstances, which will send a message to the client such that
	 *              the client will initiate the disconnect 
	 * 
	 * instead of type label we could also use the name of the participant or the instanceID
	 */
	public void unregister( String identifier){
					// e.g. "source" OR fullname !!!
		String str;
		
		str = identifier.toLowerCase() ;
		if ((str.contentEquals("source")) || (str.contentEquals("source"))){
			unregisterByType(identifier) ;
		}else{
			unregisterByName(identifier) ;	
		}
		
	}
	public void unregister( long instanceID){
		unregisterByIID( instanceID);
	}
	
	
	public void removeRegistration( XRegistration xreg, int index ) {
		long iid;
		 
		
		// send a notification for unregistration ???
		iid = xreg.getInstanceID() ;
		
		out.print(3, "Instance (id:"+ iid +", port:"+xreg.getClientListeningPort()+") of role "+xreg.getRole()+" is going to be unregistered from MessageBoard." );
		
		metaRegistration.removeItem(index) ;
		connectionsMgr.removeConnectionItem( iid );
		
		// remove also the subscriptions !!
		if (getFutures()!=null){
			getFutures().removeAll( iid ) ;
		}
		
		showResultingSummary( iid, 19 ) ;
	}
	
	synchronized private void unregisterByType( String typelabel){
		Vector<XRegistration> registrations;
		XRegistration xreg ;
		String str;
		
		registrations = this.msgBoardShares.getMetaRegistration().getRegisteredInstances() ;
	
		// incoming new registration do not disturb, since we proceed DOWN to index 0 !
		for (int i=registrations.size()-1;i>=0;i--){
			
			xreg = registrations.get(i) ;
			str = xreg.getRole().toLowerCase() ;
			if (str.toLowerCase().contentEquals(typelabel.toLowerCase())){
				 
				removeRegistration( xreg, i );
			}
		}
		
	}

	private void unregisterByName( String clientname){
		Vector<XRegistration> registrations;
		XRegistration xreg ;
		String str;
		
		registrations = this.msgBoardShares.getMetaRegistration().getRegisteredInstances() ;
	
		// incoming new registration do not disturb, since we proceed DOWN to index 0 !
		for (int i=registrations.size()-1;i>=0;i--){
			
			xreg = registrations.get(i) ;
			str = xreg.getNameOfApplicant();
			if (str.contentEquals(clientname)){
				registrations.remove(i) ;
			}
		}
	}
	
	private void unregisterByIID( long instanceid){
		Vector<XRegistration> registrations;
		XRegistration xreg ;
		long iid;
		
		registrations = this.msgBoardShares.getMetaRegistration().getRegisteredInstances() ;
	
		// incoming new registration do not disturb, since we proceed DOWN to index 0 !
		for (int i=registrations.size()-1;i>=0;i--){
			
			xreg = registrations.get(i) ;
			iid = xreg.getInstanceID();
			if (iid == instanceid){
				registrations.remove(i) ;
			}
		}
	}

	public void setRegistrationToTested( long instanceID, boolean indicator){

		long iid;
		String remainders,name="", xmlstr="";
		int n,i,r, port;
		boolean hb;
		
		XMessage xMessage = new XMessage(out);
		XRegistration xreg;
		

		
		iid=0;n=0;
		
		for (i=0;i<metaRegistration.size();i++ ){
			xreg = metaRegistration.getItem(i) ;
			if (xreg.isInstanceTested()){
				n++;
			}
		}
		i=0;
		
		for ( i=0;i<metaRegistration.size();i++ ){	
			xreg = metaRegistration.getItem(i) ;
			
			// is this registered instance already in our collection of connections 
			iid = xreg.getInstanceID() ;
			
			if ((iid==instanceID) & (xreg!=null)){
				
				xreg.setInstanceTested(indicator);
				if (indicator==true){
					n = (metaRegistration.size()-n-1);
					if (n<=0){
						remainders = ", all confirmed.";
					}else{
						remainders = ", "+n+" remaining to be confirmed.";
					}
					out.print(2, "Connection for iid = "+iid+" ("+xreg.getRole()+") confirmed"+remainders);
					name = xreg.getNameOfApplicant() ;
					// send a final message to the participant, that it has been confirmed for the first time
					xmlstr = xMessage.createNotificationAboutConfirmedConnection( iid , name, xreg.getRole(), "BOARD" ) ;
					
					
					if ((xreg!=null) && (protocolID >= MessageBoardFactoryProperties.protocolTCP )){
						
						port = xreg.getClientListeningPort();
						
						if (port>0){
							
							NewParticipantConfirmation npc = new NewParticipantConfirmation( mb, xreg, xmlstr, port) ;
							npc.go();
							
							hb = xreg.getInstanceConnectionConfirmed();
							
						} // port ? protocol ?
					} // xreg!=null
					
					// we always set the connection to confirmed for client based protocols
					if (protocolID <= MessageBoardFactoryProperties.protocolFTP ){
						 xreg.setInstanceConnectionConfirmed(true);
					}
					
				}
				break;
			}
		}
	}

	class NewParticipantConfirmation implements Runnable{

		MessageBoard mb;
		String xmlstr;
		int port;
		XRegistration xreg ;
		
		public boolean isRunning=false;
		Thread pcThrd;
		
		
		public NewParticipantConfirmation( MessageBoard mb, XRegistration xreg, String xmlstr, int port) {
			 
			this.mb = mb;
			this.xmlstr = xmlstr;
			this.port = port;
			this.xreg = xreg;
			
			pcThrd = new Thread(this,"pconfirmThread");
		}
		public NewParticipantConfirmation create(){
			
			return this;
		}
		
		public boolean perform(){
			int r;
			boolean rB=false;
			
			// repeat several times, the attempt to connect could be too early
			SenderWrap sender = new SenderWrap(mb);
			r = sender.sendingMessage(xmlstr, port);
			
				if (r<0){
					out.printErr(3, "sendingMessage() failed in setRegistrationToTested(), code="+r);
				}else{
					rB=true;
				}
			sender = null;
			
			return rB;
		}
		
		public void go(){
			
			isRunning = true;
			
			pcThrd.start();
			
			while (isRunning){
				delay(1);
			}
		}
		
		
		public void run() {
			String str="";
			int z=0;
			int dd = 100;
			boolean hb=false ;
			 
			try{
				
				while ((isRunning) && (z<10) && (hb==false)){
					z++;
												out.printErr(4,"trying to confirm client connect ("+z+")...");
					hb = perform();
					
					if (hb==false){
						mb.out.delay(dd);
					}
					
				}
				
			}catch(Exception e){
				hb=false;
			}
			
			if (hb){
				str = "trying to confirm the connection/wiring to the participant (id:"+xreg.getInstanceID()+") has been successful." ;
				out.print(3,str);
			}else{
				str = "trying to confirm the connection/wiring to the participant (id:"+xreg.getInstanceID()+") FAILED." ;
				out.printErr(2,str);
			}
												
			xreg.setInstanceConnectionConfirmed(hb);
			
			isRunning = false;
		}
		
	}
	
	// TODO:
	// store it and try it a bit later again, creating an internal message
	// handlingDelayedOfTransaction( transactionID, -1 ) ;
	// this stores the whole transaction as a binary, creates a "message" using the 
	// filename, puts that into a queue (which is run by a separate thread)
	// the thread checks if any RECEPTOR has announced its availability
	public void handlingDelayedOfTransaction( long transactionID ){
		
	}
	/**
	 * rawXmlMsg is containing the state message
	 * 
	 * TODO: re-routing should wait a a few seconds (threaded waiting object), then we should 
	 * 
	 * 
	 * TODO: filters (FilterPackageIntf) are not correctly handled on re-routing !
	 * @param transactionID
	 * @param rawXmlMsg
	 */
	synchronized public void handlingReRoutingOfTransaction( long transactionID, long receptorID ){
		
		String 	xmlstr="" ,encodedTaskPackage,
				replacementStr="";
		int dataType=-1; // , linkRelation, relationItemID, relationGroupID
		// long ageOfExpiry;
		Transaction transaction;
		XMessage xMessage = new XMessage(out) ;
		FilterPackageIntf  filter =null;
		boolean filterIsActivated = false;
		
		XRegistration xreg;
		
		try{
			
			// RECEPTOR of iid = "receptorID" is not available... -> set it internally to busy state
			// find the XRegistration for this iid
			if (receptorID>0){
				for (int i = 0; i < metaRegistration.size(); i++) {
					xreg = metaRegistration.getItem(i);
					if (xreg.getInstanceID() == receptorID) {
						xreg.setReceptorState(XRegistration.RST_IS_RUNNING);
						break;
					}
				}
			} // is the problematic receptorID known ?
			
			out.printErr(2, "Re-routing transaction "+transactionID+"...");
			// informing the respective source
			
			// creating message, as-if it is coming from a source
			transaction = transactions.getTransactionByID(transactionID) ;
			
			transactionID = transaction.getTransactionID() ;
			
			encodedTaskPackage = (String)transaction.getDataObject();
			
			/*
			nameOfInstance = "routed source" ;
			transactionSecret = transaction.getClientSecret() ;
			contentType= transaction.getContentType() ;
			dataType= transaction.getDatatype() ;
			linkRelation = transaction.getLinkRelation() ;
			relationItemID = transaction.getRelationItemID() ;
			relationGroupID = transaction.getRelationGroupID() ;
			ageOfExpiry = transaction.getAgeOfExpiry() ;
			*/
			
			BasicMessageParams basics = new BasicMessageParams();
			
			basics.transactionID = transactionID;
			basics.instanceID = 0;
			basics.instanceName = "routed source";
			basics.privateTransactionIdentifier = transaction.getClientSecret();
			basics.roleOfSender = "SOURCE" ;
			
			DataDeliveryParamsPlus paramsplus = new DataDeliveryParamsPlus();
			
			paramsplus.ageOfExpiry = transaction.getAgeOfExpiry() ;
			paramsplus.contentType = transaction.getContentType() ;
			paramsplus.dataType = transaction.getDatatype() ;
			paramsplus.filterIncluded = (filter!=null) && (filterIsActivated) ;
			paramsplus.linkRelation =  transaction.getLinkRelation() ;
			paramsplus.relationItemID = transaction.getRelationItemID()  ;
			paramsplus.relationGroupID = transaction.getRelationGroupID()  ;
			
			
			
			xmlstr = xMessage.createDataDeliveryMessageFrame( basics , // nameOfInstance, "SOURCE", transactionID, transactionSecret,
															  paramsplus // ageOfExpiry,contentType, dataType, (filter!=null) && (filterIsActivated),linkRelation, relationItemID,relationGroupID 
															  ) ;
			
	        if (dataType==1){
	        	// replacementStr = "<resourcelocator>"+resourceLocator+"</resourcelocator>" ;
	        }else{
	        	replacementStr = "<format zip=\"0\" encoding=\"base64\"/>\n" + 
	        					 "         <values>"+encodedTaskPackage+"</values>\n" ;
	        				
	        }
			
			xmlstr = introduceDatatoMsgFrame( xmlstr, replacementStr) ;
			
			// internal delivery
			// this will add the xml to the collection of "sourceMessages" 
			handlingMsgFromSource(xmlstr);
			
			
		}catch(Exception e){
			
		}
		
	}

	public void removeTaskToAlignment( long instanceID  , long transactionID){
		
		XRegistration xreg;
		long iid;
		int sv;
		
		
		xreg = XRegistration.getItemByID( instanceID, metaRegistration.getRegisteredInstances() );
		
		if (xreg==null){
			return ;
		}
		
		sv = xreg.getReceptorState();
		if ((sv >= XRegistration.RST_IS_CLOSED) || (sv<0)){
			return;
		}
		
		iid = xreg.getInstanceID() ;
		
		if ( instancesForResultAlignment.indexOf(iid)>=0 ){
			xreg.removeTransactionToAlignment( transactionID ) ;
			
			String str = strgutil.arr2text(  xreg.getAlignmentQueue(), ";");
			out.print(3, "- - - - - - - - -  \n--- Alignment queue tid :   \n"+str+"\n- - - - - - - - -\n") ;
		}
	}
	// the values will be within XREG!! 
	// yet, the MessageBoard needs to know which iid have to be supervised
	public void putTaskToAlignment( XRegistration xreg, long transactionID){
		
		long iid;
		int sv;
		
		if (xreg==null){
			return ;
		}
		
		sv = xreg.getReceptorState();
		if ((sv >= XRegistration.RST_IS_CLOSED) || (sv<0)){
			return;
		}
		
		iid = xreg.getInstanceID() ;
		
		if ( instancesForResultAlignment.indexOf(iid)<0 ){
			instancesForResultAlignment.add(iid);
		}
		
		xreg.takeTransactionToAlignment( transactionID ) ;
		
		String str = strgutil.arr2text(  xreg.getAlignmentQueue(), ";");
		out.print(4, "- - - - - - - - -  \n+++ Alignment queue tid :   \n"+str+"\n- - - - - - - - -\n") ;
	}
	
	public boolean instanceExpectsAlignedResults( long instanceID ){
		boolean rB=false;
		
		rB = instancesForResultAlignment.indexOf(instanceID)>=0 ;
		return rB;
		
	}
	
	public void reconstructContextDescriptions(Transactions transactions){
		SourceMessageInterpreter smi ;

		smi = new SourceMessageInterpreter(this,null,out) ;
		smi.assimilateReloadedTransactions(transactions) ;
		
		smi=null;
	}
	
	public boolean transactionIsAligned( long transactionID ){
		long instanceID;
		Transaction transaction ;
		
		transaction = transactions.getTransactionByID(transactionID) ;
		instanceID = transaction.getInstanceID() ;
		
		return transactionIsAligned( instanceID,  transactionID );
		
	}
	public boolean transactionIsAligned( long instanceID,  long transactionID ){
		boolean rB = true;
		XRegistration xreg ;
		
	 		if ( instanceExpectsAlignedResults( instanceID)){
	 			
	 			xreg = XRegistration.getItemByID( instanceID, getMetaRegistration().getRegisteredInstances() );
	 			if (xreg!=null){
	 				rB = xreg.transactionIsInLine( transactionID ) ;
	 			}else{
	 				out.printErr(2, "registration not found for instance id:"+instanceID);
	 			}
	 		}
	 	return rB;
	}
	
	public void alignHandlersFor( MessageBoard secondaryMsgBoard){
		
		secondaryMsgBoard.registerHandler = this.registerHandler;
		
		secondaryMsgBoard.srcHandler = this.srcHandler;
		secondaryMsgBoard.cliHandler = this.cliHandler;
		secondaryMsgBoard.mbHandler = this.mbHandler;
		secondaryMsgBoard.inBuffHandler = this.inBuffHandler;
		secondaryMsgBoard.notifyHandler = this.notifyHandler;
		secondaryMsgBoard.taskRelayDispatcherProcess = this.taskRelayDispatcherProcess;
		secondaryMsgBoard.monitorProcess = this.monitorProcess;
		secondaryMsgBoard.cleanupCollector = this.cleanupCollector;
		secondaryMsgBoard.monitor = this.monitor;


		
		secondaryMsgBoard.connectionsMgr = this.connectionsMgr;
		secondaryMsgBoard.metaRegistration = this.metaRegistration;
		secondaryMsgBoard.registrationMgr = this.registrationMgr;
		secondaryMsgBoard.inBuffer = this.inBuffer;
		
		secondaryMsgBoard.boardMessages = this.boardMessages;
		secondaryMsgBoard.registerMessages = this.registerMessages;
		secondaryMsgBoard.waitingNotifications = this.waitingNotifications;
		secondaryMsgBoard.sourceMessages = this.sourceMessages;
		secondaryMsgBoard.receptorMessages = this.receptorMessages;
		
		
		
	}
	
	
	private String introduceDatatoMsgFrame( String xmlstr, String replacementStr){
		
		//   <!--  "##DATA##"   -->
		int pd,pc,pe ;
		String cstr;
		
		pd = xmlstr.indexOf( "##DATA##" ) ;
		pc = strgutil.indexOfBefore( xmlstr, "<!--", pd);
		pe = xmlstr.indexOf("-->", pc)+3;
		
		cstr = xmlstr.substring(pc, pe) ;
		
		xmlstr = xmlstr.replace( cstr, "  ");
		
		xmlstr = strgutil.insert( xmlstr, replacementStr, pc);
		
		
		return xmlstr;
	}

	@SuppressWarnings("unused")
	private String introduceFiltertoMsgFrame( String xmlstr, String replacementStr){
		
		//   <!--  "##DATA##"   -->
		int pd,pc,pe ;
		String cstr;
		
		pd = xmlstr.indexOf( "##FILTER##" ) ;
		pc = strgutil.indexOfBefore( xmlstr, "<!--", pd);
		pe = xmlstr.indexOf("-->", pc)+3;
		
		cstr = xmlstr.substring(pc, pe) ;
		
		xmlstr = xmlstr.replace( cstr, "  ");
		
		xmlstr = strgutil.insert( xmlstr, replacementStr, pc);
		
		
		return xmlstr;
	}
	
	@SuppressWarnings("static-access")
	protected void delay(int millis){
	
		try{
			Thread.currentThread().sleep(millis) ;
		}catch(Exception e){}
	}
	
	private int collectThreadStates(){
		
		int result = 4;
		
		
		if (srcHandler.isRunning() == false){
			result = result-1;
		}
		if (cliHandler.isRunning == false){
			result = result-1;
		}
		// if (mbHandler.isRunning == false){ result = result-1; } 
		if (inBuffHandler.isRunning == false){
			result = result-1;
		}
		
		return result;
	}


	public void showResultingSummary( long instanceID, int result){

		int ks, nr;
		String rstr, istr;

		ks = metaRegistration.countSources(2);
		nr = metaRegistration.countReceptors(2);

		rstr = "receptor";
		if (nr != 1) {
			rstr = rstr + "s";
		}

		istr = "source";
		if (ks != 1) {
			istr = istr + "s";
		}

		if ((instanceID > 0) && (result >= 0) && (result<15) ) {
			out.print(1, "New participant registered (iid:" + instanceID
					+ "), now there are " + ks + " " + istr + " and " + nr
					+ " " + rstr + ".");
		} else {
			if ((instanceID>0) && (result==19)){
				out.printErr(1, "Participant of id="+instanceID+" has been unregistered (registration timed out, or application exit), now there are "+
						 	    ks + " " + istr + " and " + nr + " " + rstr + ".");
				
			}else{
				out.print(1, "Candidate participant has not been registered, now there are "+
							 ks + " " + istr + " and " + nr + " " + rstr + ".");
			}
		}
	}


	
	public MessageBoardShareIntf getMsgBoardShares() {
		return msgBoardShares;
	}

	public ComBoxIntf getComBox() {
		return comBox;
	}


	public GlobalReceipts getGlobalReceipts() {
		return globalReceipts;
	}


	public MetaRegistration getMetaRegistration() {
		return metaRegistration;
	}


	public Transactions getTransactions() {
		return transactions;
	}


	public MsgBoardDataStorage getMsgBoardStorage() {
		return msgBoardStorage;
	}


	public ClientMatcherIntf getClientMatcher() {
		return clientMatcher;
	}


	synchronized public ReceptorLinks getReceptorLinks() {
		
		if (receptorLinks==null ){
			// this maintains a list for the assignment task->receptor and also contains 
			// the respective dispatcher
			receptorLinks = new ReceptorLinks(this,out) ;
		}
		return receptorLinks;
	}

	public ConnectionsManager getConnectionsMgr() {
		return connectionsMgr;
	}

	public MessageMatcherProcess getMsgmatcherProcess() {
		return msgmatcherProcess;
	}

	
	public Vector<XRegistration> getRegisteredInstances() {
		return metaRegistration.getRegisteredInstances() ;
	}


	public void setFuturesReference(Futures futures){
		
		this.futures = futures; 
	}

	public FuturesIntf getFutures() {
		
		if (futures==null){
			futures = mbf.getFutures() ;
		}
		return futures;
	}

	public ActiveContextHeaderDescriptions getActiveContextHeaderDescriptions() {
		return activeContextHeaderDescriptions;
	}

	public BoardMessageInterpreter getBoardMessageInterpreter() {
		return boardMessageInterpreter;
	}

	public void setDeliveryCompleted(boolean deliveryCompleted) {
		this.deliveryCompleted = deliveryCompleted;
	}


	public String getMBid(){
		return messageBoardsIdentifier;
	}

	

	public String getSimpleName() {
		return simpleName;
	}

	public XRegistration getMxReg() {
		return mxReg;
	}

	public String getConfigHomePath() {
		return configHomePath;
	}



	public void setConfigHomePath(String configHomePath) {
		this.configHomePath = configHomePath;
	}

	public int getMaxNumberOfBufferedTasks() {
		return maxNumberOfBufferedTasks;
	}


	public void setMaxNumberOfBufferedTasks(int maxNumberOfBufferedTasks) {
		this.maxNumberOfBufferedTasks = maxNumberOfBufferedTasks;
	}


	public int getDropDelayUnrelayedMessages() {
		return dropDelayUnrelayedMessages;
	}


	public void setDropDelayUnrelayedMessages(int dropDelayUnrelayedMessages) {
		this.dropDelayUnrelayedMessages = dropDelayUnrelayedMessages;
	}


	public int getFullDuplexConnection() {
		return fullDuplexConnection;
	}


	public void setFullDuplexConnection(int fullDuplexConnection) {
		this.fullDuplexConnection = fullDuplexConnection;
	}


	public long getRegistrationTimeOut() {
		return registrationTimeOut;
	}


	public void setRegistrationTimeOut(long registrationTimeOut) {
		this.registrationTimeOut = registrationTimeOut;
	}


	public int getModeForConflictingIDs() {
		return modeForConflictingIDs;
	}

	public void setModeForConflictingIDs(int modeForConflictingIDs) {
		this.modeForConflictingIDs = modeForConflictingIDs;
	}

	public void setTransactions( Transactions transacts ){
		transactions = transacts ;
	}
	
	public Vector<Message> getReceptorMessages() {
		return receptorMessages;
	}


	public Notifications getWaitingNotifications() {
		return waitingNotifications;
	}


	public int getMsgPortforSources() {
		return msgPortforSources;
	}



	public void setMsgPortforSources(int msgPortforSenders) {
		this.msgPortforSources = msgPortforSenders;
	}



	public int getMsgPortforReceptors() {
		return msgPortforReceptors;
	}



	public void setMsgPortforReceptors(int msgPortforClients) {
		this.msgPortforReceptors = msgPortforClients;
	}

	public int getMsgPortforRegistration() {
		return msgPortforRegistration;
	}

	public void setMsgPortforRegistration(int msgPortforRegistration) {
		this.msgPortforRegistration = msgPortforRegistration;
	}


	public int getMsgPortforService() {
		return msgPortforService;
	}


	public void setMsgPortforService(int msgPortforService) {
		this.msgPortforService = msgPortforService;
	}


	public boolean getVirtualPortFlag() {
		return virtualPorts;
	}

	public int getSendModeForBoard() {
		return sendModeForBoard;
	}


	public void setSendModeForBoard(int sendModeForBoard) {
		this.sendModeForBoard = sendModeForBoard;
	}


	public boolean isRestarting() {
		return restarting;
	}

	public void setRestarting(boolean restarting) {
		this.restarting = restarting;
	}

	public int getProtocolID() {
		return protocolID;
	}


	public boolean isVirtualPorts() {
		return virtualPorts;
	}


	public void setVirtualPorts(boolean virtualPorts) {
		this.virtualPorts = virtualPorts;
	}


 

	public int getLowestPortNumber() {
		return lowestPortNumber;
	}


	public void setLowestPortNumber(int lowestPortNumber) {
		this.lowestPortNumber = lowestPortNumber;
	}


	public int getHighestPortNumber() {
		return highestPortNumber;
	}


	public void setHighestPortNumber(int highestPortNumber) {
		this.highestPortNumber = highestPortNumber;
	}


	public int getMsgPortAllInOne() {
		return msgPortAllInOne;
	}


	public void setMsgPortAllInOne(int msgPortAllInOne) {
		this.msgPortAllInOne = msgPortAllInOne;
	}


	public int getMsgPortforInterBoard() {
		return msgPortforInterBoard;
	}


	public void setMsgPortforInterBoard(int msgPortforInterBoard) {
		this.msgPortforInterBoard = msgPortforInterBoard;
	}


	 
	public HostingData getHostingData() {
		return hostingData;
	}

	public String getIpAddressListStr() {
		return ipAddressListStr;
	}

	public void removeNotificationById( long transactionID){
		
		waitingNotifications.removeNotificationByTransactionID(transactionID) ;
		sConfirmation.add(transactionID) ;
	}
} // public MessageBoard{}



class RequestHandlingWrapper implements Runnable {

	MessageBoard mb;
	int scope;
	String rawXmlMsg = "";
	Thread rhThrd;
	String thrdGUID;

	public RequestHandlingWrapper(MessageBoard mb, int scope, String str, String thrdGUID) {
		this.mb = mb;
		rawXmlMsg = str;
		this.scope = scope;
		this.thrdGUID = thrdGUID;
	}

	public void go() {
		rhThrd = new Thread(this, "RequestHandlingWrapper");
		rhThrd.start();
	}

	public void run() {
		mb.out.print(5, ">>>   RequestHandlingWrapper started...");
		if (scope == 1) {
			mb.handlingRequestForRegistering(rawXmlMsg, thrdGUID);
		}

		if (scope == 2) {
			mb.handlingMsgFromSource(rawXmlMsg);
		}

		if (scope == 3) {
			mb.handlingMsgFromReceptor(rawXmlMsg);
		}
		if (scope == 4) {
			mb.handlingMsgFromBoard(rawXmlMsg);
		}
	}

}
	





class BoardMessageHandler implements Runnable{
	
	MessageBoard msgboard;
	
	boolean isRunning ;
	boolean isWorking;
	
	Vector<Message> brdMessages;
	
	private Thread thrd;
	
	public BoardMessageHandler( MessageBoard mb){
	
		msgboard = mb ;
		
		brdMessages = mb.boardMessages ;
		thrd = new Thread(this, "BoardMessageHandler");
		 
		thrd.start() ;
	}
	 

	 
	
	public void run() {
		int d=0;
		isRunning = true;
		 
		
		try{
 		
			while ((isRunning) &&  ( msgboard.messageboardIsActive )) {
				
				if (isWorking == false) {
					if (msgboard.boardMessages.size() > 0) {
						d=10;
						isWorking = true;
						
						msgboard.interpreteInterBoardMsg( msgboard.boardMessages.get(0) );
						
						if (msgboard.boardMessages.size() > 0)msgboard.boardMessages.remove(0);
						
						isWorking = false;
					}
				}
				{
					if (d<=1) { msgboard.delay(5);} else{d=d-1;}
				}
				
			} // while ...
			
		}catch(Exception e){
			
		}finally{
			isRunning = false;
			isWorking = false;
		}
	}
	

	public void enforcedStop(){
		
		if (thrd != null){
			try {
				thrd.interrupt();
			} catch (Exception e) {
			}
			 
			isRunning = false;
			isWorking = false;
		}
	}

	public boolean isRunning() {
		return isRunning;
	}
} // BoardMessageHandler



class RegistrationManager implements Runnable{
	
	MessageBoard msgboard;
	
	boolean isRunning = false;
	boolean isWorking = false;

	Vector<XRegistration>  participants;
	
	Thread thrd;
	 
	
	public RegistrationManager( MessageBoard mb ){
		 
		msgboard = mb ;
		 
		thrd = new Thread(this,"RegistrationManager");
		 
		thrd.start() ;
		
		msgboard.out.print(3,"RegMgr has been started...");
	}

	  
	public void run() {
		int d=0, dt=15 ; 
		long now,iid;
		isRunning = true;
		XRegistration xreg ;
		EncounterCounter encounterCounter;
		
		encounterCounter = new EncounterCounter();
		msgboard.registrationTimeOut = 90000;
		
		participants = msgboard.getRegisteredInstances() ;
		try{

			while ((isRunning) &&  ( msgboard.messageboardIsActive )) {
				
				if (isWorking == false) {
					isWorking = true;
									msgboard.out.print(6,">>>   RegistrationManager");
									
					if (participants.size() > 0) {
						d=10;
						for (int i=0;i<participants.size();i++){
							now = System.currentTimeMillis();
							
							xreg = participants.get(i) ;
							if ( now - xreg.getLastCheckTime() > 10000 ){
								
								iid = xreg.getInstanceID();
								xreg.setLastCheckTime( System.currentTimeMillis()) ;
								 
								if ((xreg.getRegisteredSince()>0) && ( now - xreg.getRegisteredSince() > msgboard.registrationTimeOut )){
									
									if (msgboard.protocolID  <= MessageBoardFactoryProperties.protocolFTP){
										xreg.setReceptorState( XRegistration.RST_IS_TIMEDOUT);
										// despite its name, we use it for timeout control of SOURCEs too !
										int cc = encounterCounter.byID(iid) ;
msgboard.out.print(2, "iid="+iid+", cc="+cc) ;										
										if (cc<=1){
											msgboard.showResultingSummary( xreg.getInstanceID(), 19) ;
										}
										if (cc>=2){
												// xreg.setReceptorState( XRegistration.RST_IS_CLOSED);
											// remove it
											msgboard.metaRegistration.removeItemByID(iid) ;
											
											msgboard.out.printErr(2, "Instance id="+iid+" has been removed from the list of registered instances.");
										}
									} else{
										// check whether they are indeed there, using a ping, they could have died without notification ...
										
									}
								}
								if ((xreg!=null) && (xreg.getRegisteredSince()<=0)){
									xreg.setRegisteredSince(now);
								}
							}
						
						} // i>
					
					} // any ?
					isWorking = false;
					
				} // not working ??
				{
					if ( d>10){
						d--;
						msgboard.delay(5) ;
					}else{
						msgboard.delay(dt);  
					}
				}
				
			} // while ...
			
		}catch(Exception e){
			e.printStackTrace(); 
		}finally{
			isRunning = false;
			isWorking = false;
		}
	}
	
	public void enforcedStop(){
	
		if (thrd != null){
			try {
				thrd.interrupt();
			} catch (Exception e) {
			}
			
			
			isRunning = false;
			isWorking = false;
		}
	}


	public boolean isRunning() {
		return isRunning;
	}
	
	public void startHandler(){
		thrd.start() ;
	}
	
}

class RegisterHandler extends MessageHandlerAbs {
	
	
	// move as much as possible to the abstract class
	public RegisterHandler( MessageBoard mb ){
		super(mb,"RegisterHandler");
		
		msgboard.out.print(3,"srchandler has been started...");
	}

	 
	
	public void run() {
		int d=0; 
		isRunning = true;
		
		try{

			while ((isRunning) &&  ( msgboard.messageboardIsActive )) {
				
				if (isWorking == false) {
					isWorking = true;
									msgboard.out.print(6,">>>   RegisterHandler");
					if (msgboard.registerMessages.size() > 0) {
						d=10;
									msgboard.out.print(5,"RegisterHandler-process : now working on first of "+msgboard.registerMessages.size()+" messages ...");
									
						msgboard.interpreteRegistryMessages( msgboard.registerMessages.get(0) );
					
						
						if (msgboard.registerMessages.size() > 0)msgboard.registerMessages.remove(0);
					
					
					}
					isWorking = false;
				} 
				{
					if (d<=1) { msgboard.delay(15);} else{d=d-1;}
				}
				
			} // while ...
			
		}catch(Exception e){
			e.printStackTrace(); 
		}finally{
			isRunning = false;
			isWorking = false;
		}
	}
	
	public void enforcedStop(){
	
		if (thrd != null){
			try {
				thrd.interrupt();
			} catch (Exception e) {
			}
			
			
			isRunning = false;
			isWorking = false;
		}
	}


	public boolean isRunning() {
		return isRunning;
	}
	
	public void startHandler(){
		thrd.start() ;
	}
	
} // RegisterHandler


class SourceHandler extends MessageHandlerAbs {
	
	
	// move as much as possible to the abstract class
	public SourceHandler( MessageBoard mb ){
		super(mb,"SourceHandler");
		
		msgboard.out.print(3,"srchandler has been started...");
		thrd.setPriority( Thread.NORM_PRIORITY + 2);
	}

	 
	
	public void run() {
		int d=0; 
		isRunning = true;
		
		try{

			while ((isRunning) &&  ( msgboard.messageboardIsActive )) {
				
				if (isWorking == false) {
					isWorking = true;
												msgboard.out.print(6,">>>   SourceHandler");
					if (msgboard.sourceMessages.size() > 0) {
						d=10;
									msgboard.out.print(6,"srchandling process : now working on first of "+msgboard.sourceMessages.size()+" messages ...");
									
						msgboard.interpreteSourceMsg( msgboard.sourceMessages.get(0) );
					
						
						if (msgboard.sourceMessages.size() > 0)msgboard.sourceMessages.remove(0);
					
					
					}
					isWorking = false;
				}  
				{
					if (d<=1) { msgboard.delay(5);} else{d=d-1;}
				}
				
			} // while ...
			
		}catch(Exception e){
			e.printStackTrace(); 
		}finally{
			isRunning = false;
			isWorking = false;
		}
	}
	 
	
} // SourceHandler




class ClientHandler implements Runnable{
	
	MessageBoard msgboard;
	
	boolean isRunning = false;
	boolean isWorking = false;
	
	// Vector<Message> receptMessages;
	private Thread thrd;
	 
	
	public ClientHandler( MessageBoard mb){
	
		msgboard = mb ;
		
		// msgboard.receptorMessages = mb.receptorMessages ;
		
		thrd = new Thread(this,"ClientHandler");
		
		thrd.start() ;
	}

	 

	 
	
	public void run() {
		int d=0;
		isRunning = true;
		
		try{

			while ((isRunning) &&  ( msgboard.messageboardIsActive )) {
				
				if (isWorking == false) {
					if (msgboard.receptorMessages.size() > 0) {
						d=10;
						isWorking = true;
						
						msgboard.interpreteReceptorMsg(msgboard.receptorMessages.get(0));
						
						isWorking = false;
						msgboard.receptorMessages.remove(0);
						
					}
				} 
				{
					if (d<=1) { msgboard.delay(5);} else{d=d-1;}
				}
				
			} // while ...
			
			
		}catch(Exception e){
			
		}finally{
			isRunning = false;
			isWorking = false;
		}
	}
	
	public void enforcedStop(){
		
		if (thrd != null){
			try {
				thrd.interrupt();
			} catch (Exception e) {
			}
			isRunning = false;
			isWorking = false;
		}
	}

	public boolean isRunning() {
		return isRunning;
	}
} // ClientHandler

 
//   -> deliveredResultsTID

/**
 * works on the collection of notifications (transactionID, xRegistrations )
 * 
 */
class NotificationHandler implements Runnable{

	MessageBoard msgboard;
	
	boolean isRunning = false;
	boolean isWorking = false;
	 
	private Thread thrd;
	 
	
	public NotificationHandler( MessageBoard mb ){
	
		msgboard = mb ;
		  
		thrd = new Thread(this,"NotesHandler");
		 
		thrd.setPriority( Thread.NORM_PRIORITY + 1);
		
		thrd.start() ;
	}

	 

	 
	
	@SuppressWarnings("unused")
	public void run() {
		isRunning = true;
		int dd,d=0,n  ;
		long now, lastCheckingTime, nid=0,tid,id ;
		Notification noty ;
		Transaction t;
		boolean timeOk,isAligned;
		
		
		try{

			while ((isRunning) &&  ( msgboard.messageboardIsActive )) {
				//cycleStartedAt = System.currentTimeMillis() ;
				
				if (isWorking == false) {
					
					if (msgboard.DEBUG==true){
						dd=8000;
					}else{
						dd=1;
					}
					 
					msgboard.out.print(7,">>>   waitingNotifications n="+msgboard.waitingNotifications.size()) ;
					
					if (msgboard.waitingNotifications.size() > 0) {
						d=10;
						n=msgboard.waitingNotifications.size();
												msgboard.out.print(6,">>>   number of notifications waiting : "+n);
						isWorking = true;	
						nid=0; 
						
						int w=0;
						while (msgboard.waitingNotifications.size()>0){
							 
							// handlingSingleNotificationAboutResults() may set it to 1, indicating to bypass any notification
							// because the notifications / results have to be aligned !
							// it also resets it to 0 after handling the aligned result sending
							if (msgboard.notificationIdlingValue>0){
								if (msgboard.notificationIdlingValue==1){
									msgboard.notificationIdlingValue=2;
								}
								msgboard.out.delay(1);
								continue;
							}
							 	
							
							w++;
							now = System.currentTimeMillis();
												if (w%20==0){
													msgboard.out.print(4,"handling internal notifications about results, currently "+msgboard.waitingNotifications.size()+" items waiting...");
													if (w>101){w=1;}
												}
							noty = msgboard.waitingNotifications.get(0) ;
							
							lastCheckingTime = noty.getLastEventTime();
							timeOk = (System.currentTimeMillis() - lastCheckingTime) > 5000;
											tid = noty.getResultsOfTransactionID();
											
							isAligned = msgboard.transactionIsAligned( tid );
							
							if ((timeOk) && (noty.isByPassProcess()==false) && (noty.getNotificationProcessRunning()==false) && (isAligned)){
								noty.setNotificationProcessRunning(true) ;
								// call the treatment on the particular notifications
								// this treatment decides, whether to send, and also
								// it removes items from the list
								// just a remove, but synchronized
								
								// check whether the the instance is here ... if not we need not to issue a Notification...
								// TODO needs the iid as direct part without additional reference to the transaction
								msgboard.waitingNotifications.removeItem(0);
								
								nid = noty.getNotificationID();
								t = msgboard.transactions.getTransactionByID(nid) ;
												msgboard.out.print(2,"going to handle internal notification task, id="+nid+", state of transaction : "+t.getTransactState()+" ("+Transaction.decodeTransactionCode(t.getTransactState())+")  "); 
									// for handling this notification, we will open a further thread
								msgboard.handlingNotification(now, noty);

												msgboard.out.print(4,"size of queue of waiting notifications after handling : "+ msgboard.waitingNotifications.size());
										
								noty.setLastEventTime(System.currentTimeMillis()) ;
								noty.setNotificationProcessRunning(false) ;
								// if it fails, that process will add the notification again to this list of waitingNotifications
								// where ? handlingNotification()?
							} // timeOk ?
							
							msgboard.delay(5);
							// check whether there are zombie?? notifications : unnecessary, since the transaction has already been delivered
							cleanupNotifications();
						} // size()>0
						
					
					} // size() > 0 ?
					isWorking = false;
				}
				{
					if (d<=1) { msgboard.delay(5);} else{d=d-1;} 
				}
				
			} // while ...
			
		}catch(Exception e){
			
		}finally{
			isRunning = false;
			isWorking = false;
		}
	}
	
	private void cleanupNotifications(){
		Notification noty;
		long tid ;
		int ix, sv;
		Transaction transaction ;
		
		int i=0;
		
		while ((i<msgboard.waitingNotifications.size()) && (msgboard.waitingNotifications.size()>0)){
			
			noty = msgboard.waitingNotifications.get(i) ;
			if (noty.getNotificationProcessRunning()==false){
				tid = noty.getNotificationID() ;
				if (tid>=0){
					ix = msgboard.getTransactions().getIndexByTransactionID(tid, 1) ;
					if (ix>=0){
						transaction = msgboard.getTransactions().getItem(ix) ;
						sv = transaction.getTransactState();
						if ((sv>=Transaction.TS_RESULTS_RETURNED) || (sv<0)){
							noty=null;
							msgboard.waitingNotifications.removeItem(i) ;
							i--;
						}
					}
				}
			}
			i++;
		} // i->
		
	}
	
	public void enforcedStop(){
		
		if (thrd != null){
			try {
				thrd.interrupt();
			} catch (Exception e) {
			}
			isRunning = false;
			isWorking = false;
		}
	}

	public boolean isRunning() {
		return isRunning;
	}
	
	 
} // NotificationHandler


class InBufferHandler implements Runnable{
	
	MessageBoard msgboard;
	
	public boolean isRunning = false;
	boolean isWorking = false;
	
	// Vector<Object> ibMessages;
	private Thread thrd;
	 
	
	public InBufferHandler( MessageBoard mb){
	
		msgboard = mb ;
		
		// ibMessages = mb.inBuffer  ;
		
		thrd = new Thread(this,"InBufferHandler");
		thrd.setName("InBufferHandler");
		thrd.start() ;
	}

	 

	 
	public void run() {
		int d=0;
		isRunning = true;
		String msg;

		try{

			while ((isRunning) &&  ( msgboard.messageboardIsActive )) {
				
				//thrd.sleep(200); System.out.print(" +");
				
				if ( msgboard.inBuffer.size() > 0){
												msgboard.out.print(5,"   ... inBuffer treatment Z : "+ msgboard.inBuffer.size());
				}
				
				if (isWorking == false) {
					msgboard.messageDispatching = false;
												msgboard.out.print(7,">>>   InBufferHandler");
					while ( msgboard.inBuffer.size() > 0)   {
						d=10;
						isWorking = true;
						try {
												msgboard.out.print(6,"   ... inBuffer treatment A (get msg, route msg, remove item -> remaining:"+ msgboard.inBuffer.size()+")...");
							msg = (String)(msgboard.inBuffer.get(0));
							msgboard.inBuffer.remove(0);
							
							if (msgboard.inBuffer.size() == 0){
								isWorking = false; 
							}
							msgboard.messageDispatcher(msg);
												msgboard.out.print(6,"   ... inBuffer treatment B (remaining:"+ msgboard.inBuffer.size()+")...");
							
							
						} catch (Exception e) {
							// educated silence...
						}
						
					} 
					{
						if (d<=1) { msgboard.delay(5);} else{d=d-1;}
					}
					if (msgboard.out.getPrintlevel()>=5){ msgboard.delay(500); }
					isWorking = false; 
				}
			} // while ...
			
		}catch(Exception e){
			e.printStackTrace();
		}finally{
			isRunning = false;
			isWorking = false;
		}
		msgboard.out.print(4,">>>>>>>>>>>>>> MessageBoard inBuffer has been stopped.");
	}
	
	public void enforcedStop(){
		
		if (thrd != null){
			try {
				thrd.interrupt();
			} catch (Exception e) {
			}
			isRunning = false;
			isWorking = false;
		}
	}

	public boolean isRunning() {
		return isRunning;
	}




	public Thread getThrd() {
		return thrd;
	}
} // InBufferHandler


class TaskRelayDispatcherProcess implements Runnable{
	
	MessageBoard msgboard;
	Thread trdThrd;
	
	boolean isRunning = false, isWorking=false;
	
	
	
	
	public TaskRelayDispatcherProcess( MessageBoard mb){
		this.msgboard = mb ;
		go();
	}

	 
	public void go(){
		trdThrd = new Thread(this,"TaskRelayDispatcherProcess"); 
		trdThrd.start();
	}
	

	
	public void run() {
		int d=0;
		isRunning = true;
		 
		if (msgboard.periodLengthForTaskDispatcher<500){
			msgboard.periodLengthForTaskDispatcher=200;
		}
		try{
		
			while ((isRunning) &&  ( msgboard.messageboardIsActive )) {
				
				if (isWorking == false) {
					// if (msgboard.boardMessages.size() > 0) 
					{
						d=10;
						isWorking = true;
						
						try{
							
							msgboard.performTaskDispatch() ;
							
						}catch(Exception e){
							msgboard.out.printErr(2, "problem when calling <performTaskDispatch()>, msg = "+e.getMessage()) ;
						}
						  
						msgboard.delay( msgboard.periodLengthForTaskDispatcher) ;
						isWorking = false;
					}
				}
				 
				
			} // while ...
			
		}catch(Exception e){
			
		}finally{
			isRunning = false;
			isWorking = false;
		}
	}
	

	public void enforcedStop(){
		
		if (trdThrd != null){
			try {
				trdThrd.interrupt();
			} catch (Exception e) {
			}
			 
			isRunning = false;
			isWorking = false;
		}
	}

	public boolean isRunning() {
		return isRunning;
	}
	
	
	
}

class TaskItemsCleanupCollector extends MessageHandlerAbs {
	

	MessageBoard mb ;
	 
	
	// move as much as possible to the abstract class
	public TaskItemsCleanupCollector( MessageBoard mb ){
		super(mb,"TaskItemsCleanupCollector");
		
		this.mb = mb ;
		
		msgboard.out.print(3,"MonitorProcess has been started...");
	}

	 
	private void checkForLeftOverTransactions(){
		
	}
	
	public void run() {
		int d=0; 
		isRunning = true;
		
		try{

			while ((isRunning) &&  ( msgboard.messageboardIsActive )) {
				
				if (isWorking == false) {
					isWorking = true;
					
					
					msgboard.out.print(6,">>>   RegisterHandler");

					// comparing the activity state of RECEPTORS and the state flags of transactions...
					
					checkForLeftOverTransactions();
					 					
									
					isWorking = false;
				} 
				
				msgboard.delay(15000);
				
				
			} // while ...
			
		}catch(Exception e){
			e.printStackTrace(); 
		}finally{
			isRunning = false;
			isWorking = false;
		}
	}
	
	public void enforcedStop(){
	
		if (thrd != null){
			try {
				thrd.interrupt();
			} catch (Exception e) {
			}
			
			
			isRunning = false;
			isWorking = false;
		}
	}


	public boolean isRunning() {
		return isRunning;
	}
	
	public void startHandler(){
		thrd.start() ;
	}
	
} // Monitor



class MonitorProcess extends MessageHandlerAbs {
	
	MessageBoard mb ;
	MessageBoardTrafficMonitorIntf monitor;
	
	DelayFor xDelay ;
	
	// move as much as possible to the abstract class
	public MonitorProcess( MessageBoard mb ,  MessageBoardTrafficMonitorIntf monitor){
		super(mb,"MonitorProcess");
		
		this.mb = mb ;
		this.monitor = monitor ;
		
		msgboard.out.print(3,"MonitorProcess has been started...");
		xDelay = (new DelayFor());
	}

	 
	
	public void run() {
		int d=0,z=0; 
		isRunning = true;
		
		try{

			while ((isRunning) &&  ( msgboard.messageboardIsActive )) {
				
				if (isWorking == false) {
					isWorking = true;
					
					
					msgboard.out.print(6,">>>   RegisterHandler");

					

					if ((monitor != null)  ){
					
						// should be its own small process ....
						// on matching criteria (e.g. every nth message, after x seconds) this performs
						// a re-calculation of the load and informs the MessageBoard/Factory about the load 
						boolean hb = monitor.checkMonitors();
			 
						hb = true;
						z++;
					
								if (hb){
									 DecimalFormat deci = new DecimalFormat("#.##");
								        
									 MonitorStatistics  stats = monitor.getStats();
									 stats.recalc();
									 
									 String vs1 = deci.format( stats.getRatePerSecond() );
									 String vs2 = deci.format( stats.getBytesPerSeconds()/1000 );
									 
									 double vc = monitor.timeSinceStart("s",1);
									 if (stats.getRatePerSecond()>0.05){
										 // outx mb.out.print(3, "average rates, msg/sec : "+vs1+" , kbytes/sec : "+vs2+" ");	 
									 }
									 // mb.out.print(2, "average rates, msg: "+vs1+" ,  bytes: "+vs2+" ");
								 }
					}
									
					isWorking = false;
				} 
				if (xDelay!=null){
					xDelay.period(10000, 100);
				} else{
					xDelay = (new DelayFor());
				}
				
			} // while ...
			
		}catch(Exception e){
			e.printStackTrace(); 
		}finally{
			isRunning = false;
			isWorking = false;
			xDelay.setUserbreak(true) ;
		}
	}
	
	public void enforcedStop(){
	
		if (thrd != null){
			try {
				thrd.interrupt();
			} catch (Exception e) {
			}
			
			
			isRunning = false;
			isWorking = false;
		}
	}


	public boolean isRunning() {
		return isRunning;
	}
	
	public void startHandler(){
		thrd.start() ;
	}
	
} // Monitor


